Add file watcher integration and IDE enhancements

- Implement file system watching with watchdog for real-time file panel updates
- Add FileEventMessage for event-driven file change handling
- Integrate Claude events parsing for IDE awareness of Claude actions
- Add extension hooks for file changes (clide_on_file_changed, clide_on_file_saved)
- Enhance editor pane with syntax highlighting service
- Improve workspace panel with action bar and file operations
- Update Claude panel with PTY terminal integration
- Add TODO.md to track long-term project items
- Update keybindings to use Alt-based shortcuts

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Jeroen Schweitzer
2026-02-01 16:14:44 +01:00
co-authored by Claude Opus 4.5
parent cd7f672638
commit c955e3ad25
27 changed files with 4275 additions and 104 deletions
+17 -14
View File
@@ -9,7 +9,7 @@ Clide is a TUI IDE wrapper for Claude Code CLI, designed to be Claude-centric wi
### Design Principles
- **Claude-centric**: Claude Code is the primary workspace, always visible
- **Contextual panels**: Editor/Diff/Terminal appear only when needed
- **VSCode-familiar**: Keybindings follow VSCode conventions
- **Alt-key shortcuts**: Alt-based keybindings don't interfere with input fields
- **Responsive**: Works on 13" laptop to widescreen monitors
- **State preservation**: Hiding panels preserves all state (never destroy widgets)
@@ -246,22 +246,25 @@ error = "#f44747"
- Settings: `theme = "summer-night"` in ClideSettings
- Runtime: `app.theme = "dracula"`
## Keybindings (VSCode-style)
## Keybindings (Alt-based)
| Action | Binding |
|--------|---------|
| Command palette | `Ctrl+Shift+P` |
| Quick open | `Ctrl+P` |
| Toggle left sidebar | `Ctrl+B` |
| Toggle right sidebar | `Ctrl+Shift+B` |
| Toggle terminal | `` Ctrl+` `` |
| Focus Claude | `Ctrl+1` |
| Focus Editor | `Ctrl+2` |
| Focus Terminal | `Ctrl+3` |
| Toggle compact mode | `Ctrl+Shift+C` |
| Git panel | `Ctrl+Shift+G` |
| Problems panel | `Ctrl+Shift+M` |
| Select theme | `Ctrl+K Ctrl+T` |
| Quit | `Alt+Q` |
| Command palette | `Alt+P` |
| Quick open | `Alt+O` |
| Toggle left sidebar | `Alt+B` |
| Toggle right sidebar | `Alt+Shift+B` |
| Toggle terminal | `` Alt+` `` |
| Focus Claude | `Alt+1` |
| Focus Editor | `Alt+2` |
| Focus Terminal | `Alt+3` |
| Toggle compact mode | `Alt+C` |
| Git panel | `Alt+G` |
| Problems panel | `Alt+M` |
| Select theme | `Alt+T` |
| Save file | `Alt+S` |
| Go to line | `Alt+L` |
## Configuration
+104
View File
@@ -0,0 +1,104 @@
# TODO
Long-term open items for Clide development.
## Core Features
### Claude Integration
- [ ] Streaming markdown responses in Claude panel
- [ ] Visual distinction between Claude responses, tool calls, and user input
- [ ] Claude history browser (past conversations)
- [ ] Claude diff flow (propose changes → diff tab → accept/reject)
### Editor
- [ ] Multi-file tab support with state preservation
- [ ] Cursor position and scroll position persistence
- [ ] Undo/redo history preservation when hiding panel
- [ ] Find in file (`Ctrl+F`)
- [ ] Go to line (`Ctrl+G`)
### Diff Panel
- [ ] Side-by-side diff view
- [ ] Unified diff view toggle
- [ ] Accept/Reject buttons for Claude-proposed changes
- [ ] Syntax highlighting in diff content
### Terminal
- [ ] Full PTY integration for terminal emulation
- [ ] Command history preservation
- [ ] Output buffer retention when hiding
### Git Integration
- [ ] Stage/unstage files from Git tab
- [ ] Discard changes context menu
- [ ] Git graph visualization (Tree tab)
- [ ] Branch popout with checkout/new branch actions
## Context Panel (Right Sidebar)
### Problems View
- [ ] Linter integration (ruff, eslint, etc.)
- [ ] Click to navigate to file:line
- [ ] Reactive problem count badge
### TODOs View
- [ ] Scan for TODO/FIXME/HACK/XXX comments
- [ ] Click to navigate to file:line
- [ ] Reactive count badge
### Jira View
- [ ] Render Jira CLI markdown output
- [ ] Auto-refresh on panel focus
- [ ] Configurable refresh interval
## UI/UX
### Responsiveness
- [ ] CSS breakpoints for different terminal widths
- [ ] Auto-hide sidebars on narrow terminals (<100 cols)
- [ ] Compact mode toggle (`Alt+C`)
### Fullscreen Mode
- [ ] Any panel can go fullscreen (`F11`)
- [ ] Exit fullscreen with `Escape`
### Command Palette
- [ ] Implement command palette (`Alt+P`)
- [ ] Quick open file (`Alt+O`)
## State Management
### Session Persistence
- [ ] Remember open files across sessions
- [ ] Persist panel sizes and layout
- [ ] Save last git state
- [ ] Remember expanded/collapsed sections
### Multiple Projects
- [ ] Workspace switcher
- [ ] Recent projects list
## Plugin System
- [ ] User-defined panels via pluggy
- [ ] Custom integrations support
- [ ] Extension API documentation
## Testing
- [ ] Snapshot tests for all panels
- [ ] Integration tests for panel communication
- [ ] Unit tests for controllers
- [ ] Unit tests for services
## Documentation
- [ ] User guide
- [ ] Plugin development guide
- [ ] Architecture documentation updates
## Build & Distribution
- [ ] PyInstaller builds for macOS
- [ ] PyInstaller builds for Linux
- [ ] CI/CD pipeline with Gitea Actions
+215 -22
View File
@@ -17,7 +17,16 @@ from clide.controllers.problems import ProblemsController
from clide.controllers.todos import TodosController
from clide.extensions.manager import ExtensionManager
from clide.models.config import ClideSettings
from clide.services.claude_events import (
ClaudeEvent,
FileReadEvent,
FileEditEvent,
FileWriteEvent,
setup_event_parsing,
)
from clide.services.file_watcher import FileEvent, FileEventMessage, setup_file_watching
from clide.services.settings_service import SettingsService, get_settings_service
from clide.services.syntax_service import register_languages
from clide.themes.registry import get_all_themes, get_theme
from clide.widgets.panels.claude import ClaudePanel
from clide.widgets.panels.context import ContextPanel
@@ -62,11 +71,13 @@ class ClideApp(App[None]):
SidebarPanel {
width: 20%;
min-width: 25;
max-width: 50;
}
ContextPanel {
width: 25%;
min-width: 30;
max-width: 50;
}
/* Workspace + Claude layout */
@@ -104,35 +115,41 @@ class ClideApp(App[None]):
}
"""
# VSCode-style keybindings
# Keybindings
# OS-native shortcuts (Ctrl on Windows/Linux, Cmd on Mac mapped to ctrl in terminal)
# Alt-based shortcuts for actions that shouldn't interfere with input fields
# Note: priority=True ensures bindings work even when widgets have focus
BINDINGS: ClassVar[list[Binding]] = [
# Global
Binding("ctrl+q", "quit", "Quit"),
Binding("ctrl+shift+p", "command_palette", "Commands"),
Binding("ctrl+p", "quick_open", "Quick Open"),
Binding("ctrl+b", "toggle_sidebar", "Toggle Sidebar"),
Binding("ctrl+shift+b", "toggle_context", "Toggle Context"),
Binding("ctrl+`", "toggle_terminal", "Toggle Terminal"),
Binding("ctrl+shift+c", "toggle_compact", "Compact Mode"),
Binding("alt+q", "quit", "Quit", priority=True),
Binding("alt+p", "command_palette", "Commands"),
Binding("alt+o", "quick_open", "Quick Open"),
Binding("alt+b", "toggle_sidebar", "Sidebar", priority=True),
Binding("alt+shift+b", "toggle_context", "Context", priority=True),
Binding("alt+`", "toggle_terminal", "Terminal", priority=True),
Binding("alt+c", "toggle_compact", "Compact", priority=True),
Binding("f11", "toggle_fullscreen", "Fullscreen"),
Binding("escape", "escape", "Escape", show=False),
# Navigation
Binding("ctrl+1", "focus_claude", "Focus Claude", show=False),
Binding("ctrl+2", "focus_editor", "Focus Editor", show=False),
Binding("ctrl+3", "focus_terminal", "Focus Terminal", show=False),
Binding("ctrl+0", "focus_sidebar", "Focus Sidebar", show=False),
Binding("alt+1", "focus_claude", "Claude", show=False, priority=True),
Binding("alt+2", "focus_editor", "Editor", show=False, priority=True),
Binding("alt+3", "focus_terminal", "Terminal", show=False, priority=True),
Binding("alt+0", "focus_sidebar", "Sidebar", show=False, priority=True),
Binding("alt+w", "close_tab", "Close Tab", show=False),
Binding("ctrl+w", "close_tab", "Close Tab", show=False),
# Git
Binding("ctrl+shift+g", "open_git", "Git", show=False),
Binding("alt+g", "open_git", "Git", show=False, priority=True),
# Problems
Binding("ctrl+shift+m", "open_problems", "Problems", show=False),
Binding("alt+m", "open_problems", "Problems", show=False, priority=True),
Binding("f8", "next_problem", "Next Problem", show=False),
Binding("shift+f8", "prev_problem", "Prev Problem", show=False),
# Editor
Binding("ctrl+s", "save_file", "Save", show=False),
Binding("ctrl+g", "goto_line", "Go to Line", show=False),
# Editor - OS-native shortcuts with priority
Binding("ctrl+s", "save_file", "Save", priority=True),
Binding("alt+s", "save_file", "Save", show=False, priority=True),
Binding("ctrl+z", "undo", "Undo", show=False, priority=True),
Binding("alt+l", "goto_line", "Go to Line", show=False),
# Theme
Binding("ctrl+k ctrl+t", "select_theme", "Select Theme", show=False),
Binding("alt+t", "select_theme", "Theme", priority=True),
]
# Reactive state
@@ -172,6 +189,9 @@ class ClideApp(App[None]):
# Register themes
self._register_themes()
# Register additional syntax highlighting languages
register_languages()
def _register_themes(self) -> None:
"""Register all themes with Textual."""
for theme_meta in get_all_themes():
@@ -192,6 +212,7 @@ class ClideApp(App[None]):
self.theme = theme_name
if save:
self._settings_service.set("theme", theme_name)
self.notify(f"Theme set to: {theme_name}", severity="information")
def save_user_settings(self) -> None:
"""Save current user settings to disk."""
@@ -234,6 +255,12 @@ class ClideApp(App[None]):
self.extension_manager.load_extensions()
await self.extension_manager.trigger_app_startup(self)
# Set up file watching for real-time sync
self._setup_file_watching()
# Set up Claude event parsing for IDE integration
self._setup_claude_events()
# Initial data refresh
await self._refresh_git()
await self._refresh_problems()
@@ -244,6 +271,113 @@ class ClideApp(App[None]):
# Focus Claude panel
self.action_focus_claude()
def _setup_file_watching(self) -> None:
"""Set up file system watching for real-time updates."""
self._file_watcher = setup_file_watching(
self.workdir,
handlers=[self._on_file_changed],
)
def _setup_claude_events(self) -> None:
"""Set up Claude event parsing for IDE integration."""
setup_event_parsing(self._on_claude_event)
def _on_file_changed(self, event: FileEvent) -> None:
"""Handle file system changes by posting a FileEventMessage.
This bridges the watchdog callback to Textual's message system,
allowing widgets to subscribe to file events via on_file_event_message.
Note: This is called from the watchdog thread, so we use call_from_thread
to safely execute on the main thread. The FileEvent (Pydantic model) is
thread-safe, but we create the Message on the main thread to avoid any
potential Textual threading issues.
"""
def post_file_event():
self.post_message(FileEventMessage(event))
self.call_from_thread(post_file_event)
async def on_file_event_message(self, message: FileEventMessage) -> None:
"""Handle file event messages from the file watcher.
This is the central handler for all file system events.
The App coordinates updates to child widgets since Textual
messages bubble up (not down to children).
"""
event = message.event
# Trigger extension hooks
self.extension_manager.trigger_file_changed(event)
# Refresh file tree for created/deleted/moved files
if event.event_type in ("created", "deleted", "moved"):
try:
sidebar = self.query_one(SidebarPanel)
sidebar.refresh_files()
except Exception:
pass
# Use call_later to avoid blocking the event loop during refreshes
# This ensures UI responsiveness isn't affected by heavy git operations
if event.event_type in ("created", "modified", "deleted", "moved"):
self.call_later(self._async_refresh_after_file_change, event)
async def _async_refresh_after_file_change(self, event: FileEvent) -> None:
"""Perform async refreshes after a file change without blocking UI."""
# Refresh git status for all file changes
await self._refresh_git()
# Only refresh problems/todos for Python/text files
if event.path.suffix in (".py", ".pyi", ".txt", ".md", ".rst"):
if event.event_type in ("created", "modified"):
await self._refresh_problems()
await self._refresh_todos()
def _on_claude_event(self, event: ClaudeEvent) -> None:
"""Handle Claude Code events for IDE integration.
When Claude reads/edits files, we can update the UI accordingly.
"""
# Schedule handling on the main thread
self.call_later(self._handle_claude_event, event)
async def _handle_claude_event(self, event: ClaudeEvent) -> None:
"""Async handler for Claude events."""
if isinstance(event, FileReadEvent):
# Claude read a file - highlight in sidebar file tree
try:
sidebar = self.query_one(SidebarPanel)
sidebar.highlight_file(event.path)
except Exception:
pass
# Trigger extension hook
self.extension_manager.trigger_claude_event(
"file_read", {"path": str(event.path)}
)
elif isinstance(event, FileEditEvent):
# Claude edited a file - notify user
# Note: Git refresh is handled by FileEventMessage from the file watcher
self.notify(f"Claude edited: {event.path.name}", severity="information")
# Trigger extension hook
self.extension_manager.trigger_claude_event(
"file_edit", {"path": str(event.path)}
)
elif isinstance(event, FileWriteEvent):
# Claude created/wrote a file - notify user
# Note: Git refresh and file tree refresh are handled by
# FileEventMessage from the file watcher (event-driven)
self.notify(f"Claude wrote: {event.path.name}", severity="information")
# Trigger extension hook
self.extension_manager.trigger_claude_event(
"file_write", {"path": str(event.path)}
)
def _apply_user_settings(self) -> None:
"""Apply saved user settings on startup."""
# Panel visibility
@@ -256,6 +390,10 @@ class ClideApp(App[None]):
# Compact mode
self.compact_mode = self._user_settings.compact_mode
# Re-apply theme after mount (Textual needs this for proper initialization)
if self._user_settings.theme:
self.theme = self._user_settings.theme
# Reactive watchers
def watch_workspace_visible(self, visible: bool) -> None:
"""Update panels when workspace visibility changes."""
@@ -406,9 +544,26 @@ class ClideApp(App[None]):
pass
def action_save_file(self) -> None:
"""Save current file."""
# EditorPane handles save internally
pass
"""Save current file in editor."""
try:
workspace = self.query_one(WorkspacePanel)
if workspace.has_unsaved_changes():
workspace._action_save()
# FileSaved message will be emitted by EditorPane if successful
else:
self.notify("No unsaved changes", severity="warning")
except Exception as e:
self.notify(f"Save failed: {e}", severity="error")
def action_undo(self) -> None:
"""Undo last action in focused widget."""
# Undo is handled by the focused widget (TextArea has built-in undo)
# This action provides feedback if no undo is available
focused = self.focused
if focused and hasattr(focused, "undo"):
focused.undo()
else:
self.notify("Undo not available", severity="warning")
def action_goto_line(self) -> None:
"""Go to line dialog."""
@@ -497,9 +652,10 @@ class ClideApp(App[None]):
async def on_workspace_panel_file_saved(
self,
_event: WorkspacePanel.FileSaved,
event: WorkspacePanel.FileSaved,
) -> None:
"""Handle file save - refresh problems and git."""
self.notify(f"Saved: {event.path.name}", severity="information")
await self._refresh_git()
await self._refresh_problems()
@@ -556,3 +712,40 @@ class ClideApp(App[None]):
"""Handle Claude Code exited."""
if event.return_code != 0:
self.notify(f"Claude Code exited with code {event.return_code}", severity="warning")
def on_workspace_panel_maximize_requested(
self,
_event: WorkspacePanel.MaximizeRequested,
) -> None:
"""Handle workspace maximize request - hide sidebars."""
# Hide sidebars when workspace is maximized
sidebar = self.query_one(SidebarPanel)
context = self.query_one(ContextPanel)
sidebar.display = False
context.display = False
# Hide Claude panel
claude = self.query_one(ClaudePanel)
claude.display = False
def on_workspace_panel_restore_requested(
self,
_event: WorkspacePanel.RestoreRequested,
) -> None:
"""Handle workspace restore request - show sidebars."""
# Restore sidebars based on saved visibility settings
sidebar = self.query_one(SidebarPanel)
context = self.query_one(ContextPanel)
sidebar.display = self._user_settings.sidebar_visible
context.display = self._user_settings.context_visible
# Show Claude panel
claude = self.query_one(ClaudePanel)
claude.display = True
def on_workspace_panel_close_requested(
self,
_event: WorkspacePanel.CloseRequested,
) -> None:
"""Handle workspace close request."""
self.workspace_visible = False
+39
View File
@@ -1,5 +1,6 @@
"""Pluggy hook specifications for Clide extensions."""
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pluggy
@@ -8,6 +9,8 @@ if TYPE_CHECKING:
from textual.app import App
from textual.widget import Widget
from clide.services.file_watcher import FileEvent
hookspec = pluggy.HookspecMarker("clide")
hookimpl = pluggy.HookimplMarker("clide")
@@ -75,3 +78,39 @@ class ClideHookSpec:
Returns:
The modified (or original) widget
"""
@hookspec
def clide_on_file_changed(self, event: "FileEvent") -> None:
"""Called when a file is created, modified, deleted, or moved.
Extensions can use this to:
- Refresh TODO scanning
- Re-run linters
- Update Jira issue links
- Trigger custom actions
Args:
event: The file event with path, type, and timestamp
"""
@hookspec
def clide_on_file_saved(self, path: Path) -> None:
"""Called after a file is saved by the editor.
More specific than file_changed - only for user saves.
Args:
path: Absolute path to the saved file
"""
@hookspec
def clide_on_claude_event(self, event_type: str, data: dict[str, Any]) -> None:
"""Called when Claude Code performs an action.
Extensions can use this to react to Claude's actions,
such as opening files, making edits, or running commands.
Args:
event_type: Type of event (e.g., "file_read", "file_edit", "tool_use")
data: Event-specific data (e.g., {"path": "/path/to/file"})
"""
+29 -1
View File
@@ -1,7 +1,8 @@
"""Extension manager for loading and managing Clide extensions."""
from importlib.metadata import entry_points
from typing import TYPE_CHECKING
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pluggy
@@ -10,6 +11,8 @@ from clide.extensions.hookspecs import ClideHookSpec
if TYPE_CHECKING:
from textual.app import App
from clide.services.file_watcher import FileEvent
EXTENSION_NAMESPACE = "clide.extensions"
@@ -78,3 +81,28 @@ class ExtensionManager:
app: The Clide application instance
"""
self.hook.clide_on_app_shutdown(app=app)
def trigger_file_changed(self, event: "FileEvent") -> None:
"""Trigger file change hooks for all extensions.
Args:
event: The file event with path, type, and timestamp
"""
self.hook.clide_on_file_changed(event=event)
def trigger_file_saved(self, path: Path) -> None:
"""Trigger file saved hooks for all extensions.
Args:
path: Path to the saved file
"""
self.hook.clide_on_file_saved(path=path)
def trigger_claude_event(self, event_type: str, data: dict[str, Any]) -> None:
"""Trigger Claude event hooks for all extensions.
Args:
event_type: Type of event (e.g., "file_read", "file_edit")
data: Event-specific data
"""
self.hook.clide_on_claude_event(event_type=event_type, data=data)
+232
View File
@@ -0,0 +1,232 @@
"""Claude Code event detection and parsing.
This module provides event infrastructure for detecting Claude Code actions
from terminal output, enabling tight IDE integration.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Literal
from textual.message import Message
# Event Types
# -----------
@dataclass
class ClaudeEvent:
"""Base class for Claude Code events."""
pass
@dataclass
class FileReadEvent(ClaudeEvent):
"""Emitted when Claude reads a file."""
path: Path
@dataclass
class FileEditEvent(ClaudeEvent):
"""Emitted when Claude edits a file."""
path: Path
@dataclass
class FileWriteEvent(ClaudeEvent):
"""Emitted when Claude creates/writes a file."""
path: Path
@dataclass
class GlobEvent(ClaudeEvent):
"""Emitted when Claude searches for files."""
pattern: str
@dataclass
class GrepEvent(ClaudeEvent):
"""Emitted when Claude searches file contents."""
pattern: str
@dataclass
class ToolStartEvent(ClaudeEvent):
"""Emitted when Claude starts using a tool."""
tool_name: str
@dataclass
class ToolEndEvent(ClaudeEvent):
"""Emitted when Claude finishes using a tool."""
tool_name: str
@dataclass
class DiffProposedEvent(ClaudeEvent):
"""Emitted when Claude proposes a diff."""
content: str
# Textual Messages
# ----------------
class ClaudeEventMessage(Message):
"""Textual message wrapper for Claude events."""
def __init__(self, event: ClaudeEvent) -> None:
self.event = event
super().__init__()
# Pattern Matching
# ----------------
# Patterns for detecting Claude Code output
PATTERNS = {
# Tool invocations - Claude Code shows these with bullet points
"tool_read": re.compile(r"● Read\(([^)]+)\)"),
"tool_edit": re.compile(r"● Edit\(([^)]+)\)"),
"tool_write": re.compile(r"● Write\(([^)]+)\)"),
"tool_glob": re.compile(r"● Glob\(([^)]+)\)"),
"tool_grep": re.compile(r"● Grep\(([^)]+)\)"),
# Generic tool pattern
"tool_start": re.compile(r"● (\w+)\("),
"tool_end": re.compile(r"└─"),
# Diff headers
"diff_header": re.compile(r"^@@\s*-\d+(?:,\d+)?\s+\+\d+(?:,\d+)?\s*@@", re.MULTILINE),
"diff_file": re.compile(r"^(?:---|\+\+\+)\s+([^\s]+)", re.MULTILINE),
}
class ClaudeEventParser:
"""Parses Claude Code terminal output to detect events.
This parser is designed to work with raw terminal data fed
through the pyte event callback.
"""
def __init__(self, callback: Callable[[ClaudeEvent], None] | None = None) -> None:
"""Initialize the event parser.
Args:
callback: Optional callback invoked for each detected event.
"""
self._callback = callback
self._buffer = ""
self._current_tool: str | None = None
def set_callback(self, callback: Callable[[ClaudeEvent], None] | None) -> None:
"""Set the event callback."""
self._callback = callback
def feed(self, data: str) -> list[ClaudeEvent]:
"""Feed terminal data and return detected events.
Args:
data: Raw terminal data from Claude Code.
Returns:
List of detected events.
"""
events: list[ClaudeEvent] = []
# Add to buffer for multi-line matching
self._buffer += data
# Limit buffer size to prevent memory issues
if len(self._buffer) > 10000:
self._buffer = self._buffer[-5000:]
# Check for tool invocations
for match in PATTERNS["tool_read"].finditer(data):
path = Path(match.group(1).strip())
events.append(FileReadEvent(path=path))
for match in PATTERNS["tool_edit"].finditer(data):
path = Path(match.group(1).strip())
events.append(FileEditEvent(path=path))
for match in PATTERNS["tool_write"].finditer(data):
path = Path(match.group(1).strip())
events.append(FileWriteEvent(path=path))
for match in PATTERNS["tool_glob"].finditer(data):
pattern = match.group(1).strip()
events.append(GlobEvent(pattern=pattern))
for match in PATTERNS["tool_grep"].finditer(data):
pattern = match.group(1).strip()
events.append(GrepEvent(pattern=pattern))
# Check for generic tool start/end
for match in PATTERNS["tool_start"].finditer(data):
tool_name = match.group(1)
# Don't emit for tools we handle specifically
if tool_name not in ("Read", "Edit", "Write", "Glob", "Grep"):
events.append(ToolStartEvent(tool_name=tool_name))
self._current_tool = tool_name
if PATTERNS["tool_end"].search(data) and self._current_tool:
events.append(ToolEndEvent(tool_name=self._current_tool))
self._current_tool = None
# Check for diff content
if PATTERNS["diff_header"].search(self._buffer):
# Extract diff content (simplified - real impl would be more sophisticated)
events.append(DiffProposedEvent(content=self._buffer))
# Clear buffer after detecting diff
self._buffer = ""
# Invoke callback for each event
if self._callback:
for evt in events:
try:
self._callback(evt)
except Exception:
pass # Don't let callback errors propagate
return events
def reset(self) -> None:
"""Reset parser state."""
self._buffer = ""
self._current_tool = None
# Global parser instance for convenience
_event_parser: ClaudeEventParser | None = None
def get_event_parser() -> ClaudeEventParser:
"""Get the global event parser instance."""
global _event_parser
if _event_parser is None:
_event_parser = ClaudeEventParser()
return _event_parser
def setup_event_parsing(callback: Callable[[ClaudeEvent], None]) -> None:
"""Set up event parsing with the given callback.
This should be called during app initialization to wire up
the event parser with the terminal stream.
"""
from clide.vendor import pyte
parser = get_event_parser()
parser.set_callback(callback)
# Wire up to pyte's event callback
# The parser.feed returns events but pyte expects None return
def _feed_wrapper(data: str) -> None:
parser.feed(data)
pyte.set_event_callback(_feed_wrapper)
+116 -35
View File
@@ -3,14 +3,125 @@
from pathlib import Path
# Language extension mapping for syntax highlighting
# Maps file extensions to tree-sitter language identifiers
LANGUAGE_MAP: dict[str, str] = {
# Python
".py": "python",
".pyi": "python",
".pyw": "python",
# JavaScript/TypeScript
".js": "javascript",
".mjs": "javascript",
".cjs": "javascript",
".jsx": "javascript",
".ts": "typescript",
".tsx": "typescript",
".mts": "typescript",
".cts": "typescript",
# Web
".html": "html",
".htm": "html",
".css": "css",
".scss": "css",
".sass": "css",
".less": "css",
# Dart/Flutter
".dart": "dart",
# Data formats
".json": "json",
".jsonc": "json",
".yaml": "yaml",
".yml": "yaml",
".toml": "toml",
".xml": "xml",
# Markdown
".md": "markdown",
".markdown": "markdown",
# Shell
".sh": "bash",
".bash": "bash",
".zsh": "bash",
".fish": "bash",
# SQL
".sql": "sql",
# Other languages
".rs": "rust",
".go": "go",
".java": "java",
".c": "c",
".h": "c",
".cpp": "cpp",
".hpp": "cpp",
".cc": "cpp",
".cxx": "cpp",
".rb": "ruby",
".php": "php",
".vue": "vue",
".svelte": "svelte",
".lua": "lua",
".r": "r",
".R": "r",
".swift": "swift",
".kt": "kotlin",
".kts": "kotlin",
".scala": "scala",
".ex": "elixir",
".exs": "elixir",
}
class FileService:
"""Service for file I/O operations."""
def __init__(self, project_path: Path) -> None:
self.project_path = project_path
async def read_file(self, path: Path) -> str:
"""Read file contents.
# Static methods for simple sync operations (used by EditorPane)
@staticmethod
def read_file(path: Path) -> str:
"""Read file contents synchronously.
Args:
path: Path to file
Returns:
File contents as string
"""
return path.read_text(encoding="utf-8")
@staticmethod
def write_file(path: Path, content: str) -> bool:
"""Write content to file synchronously.
Args:
path: Path to file
content: Content to write
Returns:
True if successful
"""
try:
path.write_text(content, encoding="utf-8")
return True
except OSError:
return False
@staticmethod
def detect_language(path: Path) -> str | None:
"""Detect language from file extension.
Args:
path: File path
Returns:
Language identifier for tree-sitter or None
"""
return LANGUAGE_MAP.get(path.suffix.lower())
# Instance methods for async operations
async def read_file_async(self, path: Path) -> str:
"""Read file contents asynchronously.
Args:
path: Path to file (relative or absolute)
@@ -21,8 +132,8 @@ class FileService:
full_path = self._resolve_path(path)
return full_path.read_text(encoding="utf-8")
async def write_file(self, path: Path, content: str) -> None:
"""Write content to file.
async def write_file_async(self, path: Path, content: str) -> None:
"""Write content to file asynchronously.
Args:
path: Path to file
@@ -52,37 +163,7 @@ class FileService:
Returns:
Language identifier or None
"""
extension_map = {
".py": "python",
".js": "javascript",
".ts": "typescript",
".jsx": "jsx",
".tsx": "tsx",
".html": "html",
".css": "css",
".scss": "scss",
".json": "json",
".yaml": "yaml",
".yml": "yaml",
".toml": "toml",
".md": "markdown",
".rs": "rust",
".go": "go",
".java": "java",
".c": "c",
".cpp": "cpp",
".h": "c",
".hpp": "cpp",
".rb": "ruby",
".php": "php",
".sh": "bash",
".bash": "bash",
".sql": "sql",
".xml": "xml",
".vue": "vue",
".svelte": "svelte",
}
return extension_map.get(path.suffix.lower())
return LANGUAGE_MAP.get(path.suffix.lower())
def _resolve_path(self, path: Path) -> Path:
"""Resolve path relative to project root.
+286
View File
@@ -0,0 +1,286 @@
"""File system watching service for real-time sync.
This module provides file system monitoring capabilities for the Clide IDE,
enabling reactive updates when files change on disk.
"""
from __future__ import annotations
import asyncio
from datetime import datetime
from pathlib import Path
from typing import Callable, Literal
from pydantic import BaseModel, ConfigDict
from textual.message import Message
try:
from watchdog.observers import Observer as WatchdogObserver
from watchdog.events import (
FileSystemEventHandler as WatchdogHandler,
FileCreatedEvent,
FileModifiedEvent,
FileDeletedEvent,
FileMovedEvent,
DirCreatedEvent,
DirModifiedEvent,
DirDeletedEvent,
DirMovedEvent,
)
WATCHDOG_AVAILABLE = True
except ImportError:
WATCHDOG_AVAILABLE = False
WatchdogObserver = None # type: ignore[misc, assignment]
WatchdogHandler = object # type: ignore[misc, assignment]
FileCreatedEvent = None # type: ignore[misc, assignment]
FileModifiedEvent = None # type: ignore[misc, assignment]
FileDeletedEvent = None # type: ignore[misc, assignment]
FileMovedEvent = None # type: ignore[misc, assignment]
DirCreatedEvent = None # type: ignore[misc, assignment]
DirModifiedEvent = None # type: ignore[misc, assignment]
DirDeletedEvent = None # type: ignore[misc, assignment]
DirMovedEvent = None # type: ignore[misc, assignment]
class FileEvent(BaseModel):
"""A file system event.
Attributes:
path: The path to the file/directory that changed.
event_type: The type of change that occurred.
timestamp: When the event occurred.
is_directory: Whether this is a directory event.
old_path: For move events, the original path.
"""
model_config = ConfigDict(strict=True, frozen=True)
path: Path
event_type: Literal["created", "modified", "deleted", "moved"]
timestamp: datetime
is_directory: bool = False
old_path: Path | None = None
class FileEventMessage(Message):
"""Textual message for file events."""
def __init__(self, event: FileEvent) -> None:
self.event = event
super().__init__()
class FileWatcher:
"""Watches a directory for file system changes.
Uses watchdog for efficient cross-platform file monitoring.
Emits FileEvent objects to registered handlers.
Example:
watcher = FileWatcher(Path.cwd())
watcher.register_handler(my_handler)
watcher.start()
# ... later ...
watcher.stop()
"""
def __init__(self, root: Path, ignore_patterns: list[str] | None = None) -> None:
"""Initialize the file watcher.
Args:
root: The root directory to watch.
ignore_patterns: Glob patterns to ignore (e.g., ["*.pyc", "__pycache__"]).
"""
self._root = root.resolve()
self._ignore_patterns = ignore_patterns or [
"*.pyc",
"__pycache__",
".git",
".venv",
"venv",
"node_modules",
".mypy_cache",
".ruff_cache",
".pytest_cache",
"*.egg-info",
]
self._handlers: list[Callable[[FileEvent], None]] = []
self._observer: WatchdogObserver | None = None # type: ignore[valid-type]
self._running = False
@property
def is_available(self) -> bool:
"""Check if watchdog is available."""
return WATCHDOG_AVAILABLE
@property
def is_running(self) -> bool:
"""Check if the watcher is currently running."""
return self._running
@property
def root(self) -> Path:
"""Get the root directory being watched."""
return self._root
def register_handler(self, handler: Callable[[FileEvent], None]) -> None:
"""Register a handler for file events.
Args:
handler: A callable that accepts a FileEvent.
"""
if handler not in self._handlers:
self._handlers.append(handler)
def unregister_handler(self, handler: Callable[[FileEvent], None]) -> None:
"""Unregister a handler.
Args:
handler: The handler to remove.
"""
if handler in self._handlers:
self._handlers.remove(handler)
def _should_ignore(self, path: Path) -> bool:
"""Check if a path should be ignored based on patterns."""
path_str = str(path)
for pattern in self._ignore_patterns:
# Simple pattern matching - could be enhanced with fnmatch
if pattern.startswith("*"):
if path_str.endswith(pattern[1:]):
return True
elif pattern in path_str:
return True
return False
def _emit_event(self, event: FileEvent) -> None:
"""Emit an event to all handlers."""
if self._should_ignore(event.path):
return
for handler in self._handlers:
try:
handler(event)
except Exception:
pass # Don't let handler errors affect other handlers
def start(self) -> bool:
"""Start watching for file changes.
Returns:
True if started successfully, False if watchdog is not available.
"""
if not WATCHDOG_AVAILABLE:
return False
if self._running:
return True
event_handler = _WatchdogHandler(self)
self._observer = WatchdogObserver()
self._observer.schedule(event_handler, str(self._root), recursive=True)
self._observer.start()
self._running = True
return True
def stop(self) -> None:
"""Stop watching for file changes."""
if self._observer is not None:
self._observer.stop()
self._observer.join(timeout=5)
self._observer = None
self._running = False
class _WatchdogHandler(WatchdogHandler): # type: ignore[misc, valid-type]
"""Internal handler for watchdog events."""
def __init__(self, watcher: FileWatcher) -> None:
super().__init__()
self._watcher = watcher
def _create_event(
self,
src_path: str | bytes,
event_type: Literal["created", "modified", "deleted", "moved"],
is_directory: bool,
dest_path: str | bytes | None = None,
) -> FileEvent:
"""Create a FileEvent from watchdog event data."""
# Watchdog can return bytes or str depending on platform
src = src_path.decode() if isinstance(src_path, bytes) else src_path
dest = dest_path.decode() if isinstance(dest_path, bytes) else dest_path
return FileEvent(
path=Path(dest if dest else src),
event_type=event_type,
timestamp=datetime.now(),
is_directory=is_directory,
old_path=Path(src) if dest else None,
)
def on_created(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(
event.src_path, "created", event.is_directory
)
self._watcher._emit_event(file_event)
def on_modified(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(
event.src_path, "modified", event.is_directory
)
self._watcher._emit_event(file_event)
def on_deleted(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(
event.src_path, "deleted", event.is_directory
)
self._watcher._emit_event(file_event)
def on_moved(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(
event.src_path, "moved", event.is_directory, event.dest_path
)
self._watcher._emit_event(file_event)
# Global watcher instance
_file_watcher: FileWatcher | None = None
def get_file_watcher(root: Path | None = None) -> FileWatcher:
"""Get or create the global file watcher.
Args:
root: The root directory to watch. Only used on first call.
Returns:
The FileWatcher instance.
"""
global _file_watcher
if _file_watcher is None:
_file_watcher = FileWatcher(root or Path.cwd())
return _file_watcher
def setup_file_watching(
root: Path,
handlers: list[Callable[[FileEvent], None]] | None = None,
) -> FileWatcher:
"""Set up file watching with optional initial handlers.
Args:
root: The root directory to watch.
handlers: Optional list of handlers to register.
Returns:
The configured FileWatcher.
"""
global _file_watcher
_file_watcher = FileWatcher(root)
if handlers:
for handler in handlers:
_file_watcher.register_handler(handler)
_file_watcher.start()
return _file_watcher
+3
View File
@@ -32,6 +32,9 @@ class UserSettings(BaseModel):
# Integrations
jira_enabled: bool = False
# Debug
terminal_debug: bool = False # Verbose terminal/pyte logging to ~/.clide/terminal_debug.log
class SettingsService:
"""Service for loading and saving user settings.
+170
View File
@@ -0,0 +1,170 @@
"""Syntax highlighting service for additional language support.
Textual 7.x includes built-in support for many languages when tree-sitter
packages are installed. This module provides utilities for checking and
registering additional languages.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
pass
logger = logging.getLogger(__name__)
# Languages supported by Textual's TextArea with tree-sitter packages
SUPPORTED_LANGUAGES = {
# Core web languages
"python",
"javascript",
"typescript",
"html",
"css",
"json",
# Markup/config
"markdown",
"yaml",
"toml",
"xml",
# Shell
"bash",
# SQL
"sql",
# Systems languages
"rust",
"go",
"java",
# Regex
"regex",
}
# Additional languages that may be registered if packages are available
OPTIONAL_LANGUAGES = [
"dart",
"kotlin",
"swift",
"scala",
"ruby",
"php",
"lua",
"c",
"cpp",
"csharp",
"elixir",
"haskell",
"ocaml",
"zig",
"nim",
"vue",
"svelte",
]
def register_languages() -> list[str]:
"""Register additional languages with Textual's TextArea.
In Textual 7.x, languages are automatically registered when tree-sitter
packages are installed. This function registers additional languages
that need special handling (like TypeScript which has separate functions).
Returns:
List of successfully registered language names
"""
try:
from textual.widgets import TextArea
except ImportError:
logger.warning("Textual not available")
return []
registered = []
# Register TypeScript and TSX (they have special language function names)
try:
import tree_sitter_typescript as tst
# Register TypeScript
try:
TextArea.register_language(tst.language_typescript(), "typescript")
registered.append("typescript")
logger.debug("Registered language: typescript")
except Exception as e:
logger.debug(f"Could not register typescript: {e}")
# Register TSX
try:
TextArea.register_language(tst.language_tsx(), "tsx")
registered.append("tsx")
logger.debug("Registered language: tsx")
except Exception as e:
logger.debug(f"Could not register tsx: {e}")
except ImportError:
logger.debug("tree-sitter-typescript not installed")
# Register other optional languages with standard API
for lang_name in OPTIONAL_LANGUAGES:
try:
# Try to import the tree-sitter package for this language
module_name = f"tree_sitter_{lang_name}"
module = __import__(module_name)
# Get the language function
if hasattr(module, "language"):
language = module.language()
# Try to get a highlight query if available
highlight_query = None
if hasattr(module, "HIGHLIGHTS_QUERY"):
highlight_query = module.HIGHLIGHTS_QUERY
# Register with Textual
try:
TextArea.register_language(language, lang_name, highlight_query)
registered.append(lang_name)
logger.debug(f"Registered language: {lang_name}")
except Exception as e:
logger.debug(f"Could not register language '{lang_name}': {e}")
except ImportError:
# Package not installed, skip
pass
except Exception as e:
logger.debug(f"Error processing language '{lang_name}': {e}")
return registered
def get_available_languages() -> list[str]:
"""Get list of all available languages for syntax highlighting.
Returns:
List of language names that can be used with TextArea
"""
try:
from textual.widgets import TextArea
# Create a temporary instance to check available languages
ta = TextArea()
return sorted(ta.available_languages)
except ImportError:
return sorted(SUPPORTED_LANGUAGES)
except Exception:
return sorted(SUPPORTED_LANGUAGES)
def is_syntax_highlighting_available() -> bool:
"""Check if syntax highlighting is available.
Returns:
True if tree-sitter is installed and syntax highlighting works
"""
try:
from textual.widgets import TextArea
ta = TextArea("test", language="python")
return ta.is_syntax_aware
except Exception:
return False
+1
View File
@@ -0,0 +1 @@
"""Vendored third-party libraries for Clide."""
+19
View File
@@ -0,0 +1,19 @@
pyte - LGPL License
====================
This is a vendored copy of pyte (https://github.com/selectel/pyte)
with modifications for Clide diagnostic logging.
Original copyright:
(c) 2011-2012 by Selectel.
(c) 2012-2017 by pyte authors and contributors.
This code is licensed under the GNU Lesser General Public License (LGPL).
Modifications made by the Clide project are also licensed under LGPL.
For the full LGPL license text, see:
https://www.gnu.org/licenses/lgpl-3.0.html
Modifications:
- Added diagnostic logging hooks for debugging terminal rendering issues
- Added event callback support for Claude Code integration
+55
View File
@@ -0,0 +1,55 @@
"""
pyte
~~~~
`pyte` implements a mix of VT100, VT220 and VT520 specification,
and aims to support most of the `TERM=linux` functionality.
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
__all__ = (
"Screen", "DiffScreen", "HistoryScreen", "DebugScreen",
"Stream", "ByteStream",
# Clide additions
"set_debug_logger", "set_event_callback",
)
import io
from typing import Union
from .screens import Screen, DiffScreen, HistoryScreen, DebugScreen
from .screens import set_debug_logger as _set_screen_logger
from .streams import Stream, ByteStream
from .streams import set_debug_logger as _set_stream_logger
from .streams import set_event_callback
# Re-export submodules for compatibility
from . import modes
from . import screens
def set_debug_logger(logger):
"""Set debug logger for both streams and screens.
Args:
logger: A callable that accepts a string message, or None to disable.
"""
_set_stream_logger(logger)
_set_screen_logger(logger)
if __debug__:
def dis(chars: Union[bytes, str]) -> None:
"""A :func:`dis.dis` for terminals."""
if isinstance(chars, str):
chars = chars.encode("utf-8")
with io.StringIO() as buf:
ByteStream(DebugScreen(to=buf)).feed(chars)
print(buf.getvalue())
+139
View File
@@ -0,0 +1,139 @@
"""
pyte.charsets
~~~~~~~~~~~~~
This module defines ``G0`` and ``G1`` charset mappings the same way
they are defined for linux terminal, see
``linux/drivers/tty/consolemap.c`` @ http://git.kernel.org
.. note:: ``VT100_MAP`` and ``IBMPC_MAP`` were taken unchanged
from linux kernel source and therefore are licensed
under **GPL**.
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
#: Latin1.
LAT1_MAP = "".join(map(chr, range(256)))
#: VT100 graphic character set.
VT100_MAP = "".join(chr(c) for c in [
0x0000, 0x0001, 0x0002, 0x0003, 0x0004, 0x0005, 0x0006, 0x0007,
0x0008, 0x0009, 0x000a, 0x000b, 0x000c, 0x000d, 0x000e, 0x000f,
0x0010, 0x0011, 0x0012, 0x0013, 0x0014, 0x0015, 0x0016, 0x0017,
0x0018, 0x0019, 0x001a, 0x001b, 0x001c, 0x001d, 0x001e, 0x001f,
0x0020, 0x0021, 0x0022, 0x0023, 0x0024, 0x0025, 0x0026, 0x0027,
0x0028, 0x0029, 0x002a, 0x2192, 0x2190, 0x2191, 0x2193, 0x002f,
0x2588, 0x0031, 0x0032, 0x0033, 0x0034, 0x0035, 0x0036, 0x0037,
0x0038, 0x0039, 0x003a, 0x003b, 0x003c, 0x003d, 0x003e, 0x003f,
0x0040, 0x0041, 0x0042, 0x0043, 0x0044, 0x0045, 0x0046, 0x0047,
0x0048, 0x0049, 0x004a, 0x004b, 0x004c, 0x004d, 0x004e, 0x004f,
0x0050, 0x0051, 0x0052, 0x0053, 0x0054, 0x0055, 0x0056, 0x0057,
0x0058, 0x0059, 0x005a, 0x005b, 0x005c, 0x005d, 0x005e, 0x00a0,
0x25c6, 0x2592, 0x2409, 0x240c, 0x240d, 0x240a, 0x00b0, 0x00b1,
0x2591, 0x240b, 0x2518, 0x2510, 0x250c, 0x2514, 0x253c, 0x23ba,
0x23bb, 0x2500, 0x23bc, 0x23bd, 0x251c, 0x2524, 0x2534, 0x252c,
0x2502, 0x2264, 0x2265, 0x03c0, 0x2260, 0x00a3, 0x00b7, 0x007f,
0x0080, 0x0081, 0x0082, 0x0083, 0x0084, 0x0085, 0x0086, 0x0087,
0x0088, 0x0089, 0x008a, 0x008b, 0x008c, 0x008d, 0x008e, 0x008f,
0x0090, 0x0091, 0x0092, 0x0093, 0x0094, 0x0095, 0x0096, 0x0097,
0x0098, 0x0099, 0x009a, 0x009b, 0x009c, 0x009d, 0x009e, 0x009f,
0x00a0, 0x00a1, 0x00a2, 0x00a3, 0x00a4, 0x00a5, 0x00a6, 0x00a7,
0x00a8, 0x00a9, 0x00aa, 0x00ab, 0x00ac, 0x00ad, 0x00ae, 0x00af,
0x00b0, 0x00b1, 0x00b2, 0x00b3, 0x00b4, 0x00b5, 0x00b6, 0x00b7,
0x00b8, 0x00b9, 0x00ba, 0x00bb, 0x00bc, 0x00bd, 0x00be, 0x00bf,
0x00c0, 0x00c1, 0x00c2, 0x00c3, 0x00c4, 0x00c5, 0x00c6, 0x00c7,
0x00c8, 0x00c9, 0x00ca, 0x00cb, 0x00cc, 0x00cd, 0x00ce, 0x00cf,
0x00d0, 0x00d1, 0x00d2, 0x00d3, 0x00d4, 0x00d5, 0x00d6, 0x00d7,
0x00d8, 0x00d9, 0x00da, 0x00db, 0x00dc, 0x00dd, 0x00de, 0x00df,
0x00e0, 0x00e1, 0x00e2, 0x00e3, 0x00e4, 0x00e5, 0x00e6, 0x00e7,
0x00e8, 0x00e9, 0x00ea, 0x00eb, 0x00ec, 0x00ed, 0x00ee, 0x00ef,
0x00f0, 0x00f1, 0x00f2, 0x00f3, 0x00f4, 0x00f5, 0x00f6, 0x00f7,
0x00f8, 0x00f9, 0x00fa, 0x00fb, 0x00fc, 0x00fd, 0x00fe, 0x00ff
])
#: IBM Codepage 437.
IBMPC_MAP = "".join(chr(c) for c in [
0x0000, 0x263a, 0x263b, 0x2665, 0x2666, 0x2663, 0x2660, 0x2022,
0x25d8, 0x25cb, 0x25d9, 0x2642, 0x2640, 0x266a, 0x266b, 0x263c,
0x25b6, 0x25c0, 0x2195, 0x203c, 0x00b6, 0x00a7, 0x25ac, 0x21a8,
0x2191, 0x2193, 0x2192, 0x2190, 0x221f, 0x2194, 0x25b2, 0x25bc,
0x0020, 0x0021, 0x0022, 0x0023, 0x0024, 0x0025, 0x0026, 0x0027,
0x0028, 0x0029, 0x002a, 0x002b, 0x002c, 0x002d, 0x002e, 0x002f,
0x0030, 0x0031, 0x0032, 0x0033, 0x0034, 0x0035, 0x0036, 0x0037,
0x0038, 0x0039, 0x003a, 0x003b, 0x003c, 0x003d, 0x003e, 0x003f,
0x0040, 0x0041, 0x0042, 0x0043, 0x0044, 0x0045, 0x0046, 0x0047,
0x0048, 0x0049, 0x004a, 0x004b, 0x004c, 0x004d, 0x004e, 0x004f,
0x0050, 0x0051, 0x0052, 0x0053, 0x0054, 0x0055, 0x0056, 0x0057,
0x0058, 0x0059, 0x005a, 0x005b, 0x005c, 0x005d, 0x005e, 0x005f,
0x0060, 0x0061, 0x0062, 0x0063, 0x0064, 0x0065, 0x0066, 0x0067,
0x0068, 0x0069, 0x006a, 0x006b, 0x006c, 0x006d, 0x006e, 0x006f,
0x0070, 0x0071, 0x0072, 0x0073, 0x0074, 0x0075, 0x0076, 0x0077,
0x0078, 0x0079, 0x007a, 0x007b, 0x007c, 0x007d, 0x007e, 0x2302,
0x00c7, 0x00fc, 0x00e9, 0x00e2, 0x00e4, 0x00e0, 0x00e5, 0x00e7,
0x00ea, 0x00eb, 0x00e8, 0x00ef, 0x00ee, 0x00ec, 0x00c4, 0x00c5,
0x00c9, 0x00e6, 0x00c6, 0x00f4, 0x00f6, 0x00f2, 0x00fb, 0x00f9,
0x00ff, 0x00d6, 0x00dc, 0x00a2, 0x00a3, 0x00a5, 0x20a7, 0x0192,
0x00e1, 0x00ed, 0x00f3, 0x00fa, 0x00f1, 0x00d1, 0x00aa, 0x00ba,
0x00bf, 0x2310, 0x00ac, 0x00bd, 0x00bc, 0x00a1, 0x00ab, 0x00bb,
0x2591, 0x2592, 0x2593, 0x2502, 0x2524, 0x2561, 0x2562, 0x2556,
0x2555, 0x2563, 0x2551, 0x2557, 0x255d, 0x255c, 0x255b, 0x2510,
0x2514, 0x2534, 0x252c, 0x251c, 0x2500, 0x253c, 0x255e, 0x255f,
0x255a, 0x2554, 0x2569, 0x2566, 0x2560, 0x2550, 0x256c, 0x2567,
0x2568, 0x2564, 0x2565, 0x2559, 0x2558, 0x2552, 0x2553, 0x256b,
0x256a, 0x2518, 0x250c, 0x2588, 0x2584, 0x258c, 0x2590, 0x2580,
0x03b1, 0x00df, 0x0393, 0x03c0, 0x03a3, 0x03c3, 0x00b5, 0x03c4,
0x03a6, 0x0398, 0x03a9, 0x03b4, 0x221e, 0x03c6, 0x03b5, 0x2229,
0x2261, 0x00b1, 0x2265, 0x2264, 0x2320, 0x2321, 0x00f7, 0x2248,
0x00b0, 0x2219, 0x00b7, 0x221a, 0x207f, 0x00b2, 0x25a0, 0x00a0
])
#: VAX42 character set.
VAX42_MAP = "".join(chr(c) for c in [
0x0000, 0x263a, 0x263b, 0x2665, 0x2666, 0x2663, 0x2660, 0x2022,
0x25d8, 0x25cb, 0x25d9, 0x2642, 0x2640, 0x266a, 0x266b, 0x263c,
0x25b6, 0x25c0, 0x2195, 0x203c, 0x00b6, 0x00a7, 0x25ac, 0x21a8,
0x2191, 0x2193, 0x2192, 0x2190, 0x221f, 0x2194, 0x25b2, 0x25bc,
0x0020, 0x043b, 0x0022, 0x0023, 0x0024, 0x0025, 0x0026, 0x0027,
0x0028, 0x0029, 0x002a, 0x002b, 0x002c, 0x002d, 0x002e, 0x002f,
0x0030, 0x0031, 0x0032, 0x0033, 0x0034, 0x0035, 0x0036, 0x0037,
0x0038, 0x0039, 0x003a, 0x003b, 0x003c, 0x003d, 0x003e, 0x0435,
0x0040, 0x0041, 0x0042, 0x0043, 0x0044, 0x0045, 0x0046, 0x0047,
0x0048, 0x0049, 0x004a, 0x004b, 0x004c, 0x004d, 0x004e, 0x004f,
0x0050, 0x0051, 0x0052, 0x0053, 0x0054, 0x0055, 0x0056, 0x0057,
0x0058, 0x0059, 0x005a, 0x005b, 0x005c, 0x005d, 0x005e, 0x005f,
0x0060, 0x0441, 0x0062, 0x0063, 0x0064, 0x0065, 0x0066, 0x0067,
0x0435, 0x0069, 0x006a, 0x006b, 0x006c, 0x006d, 0x006e, 0x043a,
0x0070, 0x0071, 0x0442, 0x0073, 0x043b, 0x0435, 0x0076, 0x0077,
0x0078, 0x0079, 0x007a, 0x007b, 0x007c, 0x007d, 0x007e, 0x2302,
0x00c7, 0x00fc, 0x00e9, 0x00e2, 0x00e4, 0x00e0, 0x00e5, 0x00e7,
0x00ea, 0x00eb, 0x00e8, 0x00ef, 0x00ee, 0x00ec, 0x00c4, 0x00c5,
0x00c9, 0x00e6, 0x00c6, 0x00f4, 0x00f6, 0x00f2, 0x00fb, 0x00f9,
0x00ff, 0x00d6, 0x00dc, 0x00a2, 0x00a3, 0x00a5, 0x20a7, 0x0192,
0x00e1, 0x00ed, 0x00f3, 0x00fa, 0x00f1, 0x00d1, 0x00aa, 0x00ba,
0x00bf, 0x2310, 0x00ac, 0x00bd, 0x00bc, 0x00a1, 0x00ab, 0x00bb,
0x2591, 0x2592, 0x2593, 0x2502, 0x2524, 0x2561, 0x2562, 0x2556,
0x2555, 0x2563, 0x2551, 0x2557, 0x255d, 0x255c, 0x255b, 0x2510,
0x2514, 0x2534, 0x252c, 0x251c, 0x2500, 0x253c, 0x255e, 0x255f,
0x255a, 0x2554, 0x2569, 0x2566, 0x2560, 0x2550, 0x256c, 0x2567,
0x2568, 0x2564, 0x2565, 0x2559, 0x2558, 0x2552, 0x2553, 0x256b,
0x256a, 0x2518, 0x250c, 0x2588, 0x2584, 0x258c, 0x2590, 0x2580,
0x03b1, 0x00df, 0x0393, 0x03c0, 0x03a3, 0x03c3, 0x00b5, 0x03c4,
0x03a6, 0x0398, 0x03a9, 0x03b4, 0x221e, 0x03c6, 0x03b5, 0x2229,
0x2261, 0x00b1, 0x2265, 0x2264, 0x2320, 0x2321, 0x00f7, 0x2248,
0x00b0, 0x2219, 0x00b7, 0x221a, 0x207f, 0x00b2, 0x25a0, 0x00a0
])
MAPS = {
"B": LAT1_MAP,
"0": VT100_MAP,
"U": IBMPC_MAP,
"V": VAX42_MAP
}
+77
View File
@@ -0,0 +1,77 @@
"""
pyte.control
~~~~~~~~~~~~
This module defines simple control sequences, recognized by
:class:`~pyte.streams.Stream`, the set of codes here is for
``TERM=linux`` which is a superset of VT102.
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
#: *Space*: Not surprisingly -- ``" "``.
SP = " "
#: *Null*: Does nothing.
NUL = "\x00"
#: *Bell*: Beeps.
BEL = "\x07"
#: *Backspace*: Backspace one column, but not past the beginning of the
#: line.
BS = "\x08"
#: *Horizontal tab*: Move cursor to the next tab stop, or to the end
#: of the line if there is no earlier tab stop.
HT = "\x09"
#: *Linefeed*: Give a line feed, and, if :data:`pyte.modes.LNM` (new
#: line mode) is set also a carriage return.
LF = "\n"
#: *Vertical tab*: Same as :data:`LF`.
VT = "\x0b"
#: *Form feed*: Same as :data:`LF`.
FF = "\x0c"
#: *Carriage return*: Move cursor to left margin on current line.
CR = "\r"
#: *Shift out*: Activate G1 character set.
SO = "\x0e"
#: *Shift in*: Activate G0 character set.
SI = "\x0f"
#: *Cancel*: Interrupt escape sequence. If received during an escape or
#: control sequence, cancels the sequence and displays substitution
#: character.
CAN = "\x18"
#: *Substitute*: Same as :data:`CAN`.
SUB = "\x1a"
#: *Escape*: Starts an escape sequence.
ESC = "\x1b"
#: *Delete*: Is ignored.
DEL = "\x7f"
#: *Control sequence introducer*.
CSI_C0 = ESC + "["
CSI_C1 = "\x9b"
CSI = CSI_C0
#: *String terminator*.
ST_C0 = ESC + "\\"
ST_C1 = "\x9c"
ST = ST_C0
#: *Operating system command*.
OSC_C0 = ESC + "]"
OSC_C1 = "\x9d"
OSC = OSC_C0
+154
View File
@@ -0,0 +1,154 @@
"""
pyte.escape
~~~~~~~~~~~
This module defines both CSI and non-CSI escape sequences, recognized
by :class:`~pyte.streams.Stream` and subclasses.
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
#: *Reset*.
RIS = "c"
#: *Index*: Move cursor down one line in same column. If the cursor is
#: at the bottom margin, the screen performs a scroll-up.
IND = "D"
#: *Next line*: Same as :data:`pyte.control.LF`.
NEL = "E"
#: Tabulation set: Set a horizontal tab stop at cursor position.
HTS = "H"
#: *Reverse index*: Move cursor up one line in same column. If the
#: cursor is at the top margin, the screen performs a scroll-down.
RI = "M"
#: Save cursor: Save cursor position, character attribute (graphic
#: rendition), character set, and origin mode selection (see
#: :data:`DECRC`).
DECSC = "7"
#: *Restore cursor*: Restore previously saved cursor position, character
#: attribute (graphic rendition), character set, and origin mode
#: selection. If none were saved, move cursor to home position.
DECRC = "8"
# "Sharp" escape sequences.
# -------------------------
#: *Alignment display*: Fill screen with uppercase E's for testing
#: screen focus and alignment.
DECALN = "8"
# ECMA-48 CSI sequences.
# ---------------------
#: *Insert character*: Insert the indicated # of blank characters.
ICH = "@"
#: *Cursor up*: Move cursor up the indicated # of lines in same column.
#: Cursor stops at top margin.
CUU = "A"
#: *Cursor down*: Move cursor down the indicated # of lines in same
#: column. Cursor stops at bottom margin.
CUD = "B"
#: *Cursor forward*: Move cursor right the indicated # of columns.
#: Cursor stops at right margin.
CUF = "C"
#: *Cursor back*: Move cursor left the indicated # of columns. Cursor
#: stops at left margin.
CUB = "D"
#: *Cursor next line*: Move cursor down the indicated # of lines to
#: column 1.
CNL = "E"
#: *Cursor previous line*: Move cursor up the indicated # of lines to
#: column 1.
CPL = "F"
#: *Cursor horizontal align*: Move cursor to the indicated column in
#: current line.
CHA = "G"
#: *Cursor position*: Move cursor to the indicated line, column (origin
#: at ``1, 1``).
CUP = "H"
#: *Erase data* (default: from cursor to end of line).
ED = "J"
#: *Erase in line* (default: from cursor to end of line).
EL = "K"
#: *Insert line*: Insert the indicated # of blank lines, starting from
#: the current line. Lines displayed below cursor move down. Lines moved
#: past the bottom margin are lost.
IL = "L"
#: *Delete line*: Delete the indicated # of lines, starting from the
#: current line. As lines are deleted, lines displayed below cursor
#: move up. Lines added to bottom of screen have spaces with same
#: character attributes as last line move up.
DL = "M"
#: *Delete character*: Delete the indicated # of characters on the
#: current line. When character is deleted, all characters to the right
#: of cursor move left.
DCH = "P"
#: *Erase character*: Erase the indicated # of characters on the
#: current line.
ECH = "X"
#: *Horizontal position relative*: Same as :data:`CUF`.
HPR = "a"
#: *Device Attributes*.
DA = "c"
#: *Vertical position adjust*: Move cursor to the indicated line,
#: current column.
VPA = "d"
#: *Vertical position relative*: Same as :data:`CUD`.
VPR = "e"
#: *Horizontal / Vertical position*: Same as :data:`CUP`.
HVP = "f"
#: *Tabulation clear*: Clears a horizontal tab stop at cursor position.
TBC = "g"
#: *Set mode*.
SM = "h"
#: *Reset mode*.
RM = "l"
#: *Select graphics rendition*: The terminal can display the following
#: character attributes that change the character display without
#: changing the character (see :mod:`pyte.graphics`).
SGR = "m"
#: *Device status report*.
DSR = "n"
#: *Select top and bottom margins*: Selects margins, defining the
#: scrolling region; parameters are top and bottom line. If called
#: without any arguments, whole screen is used.
DECSTBM = "r"
#: *Horizontal position adjust*: Same as :data:`CHA`.
HPA = "'"
+148
View File
@@ -0,0 +1,148 @@
"""
pyte.graphics
~~~~~~~~~~~~~
This module defines graphic-related constants, mostly taken from
:manpage:`console_codes(4)` and
http://pueblo.sourceforge.net/doc/manual/ansi_color_codes.html.
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
#: A mapping of ANSI text style codes to style names, "+" means the:
#: attribute is set, "-" -- reset; example:
#:
#: >>> text[1]
#: '+bold'
#: >>> text[9]
#: '+strikethrough'
TEXT = {
1: "+bold",
3: "+italics",
4: "+underscore",
5: "+blink",
7: "+reverse",
9: "+strikethrough",
22: "-bold",
23: "-italics",
24: "-underscore",
25: "-blink",
27: "-reverse",
29: "-strikethrough",
}
#: A mapping of ANSI foreground color codes to color names.
#:
#: >>> FG_ANSI[30]
#: 'black'
#: >>> FG_ANSI[38]
#: 'default'
FG_ANSI = {
30: "black",
31: "red",
32: "green",
33: "brown",
34: "blue",
35: "magenta",
36: "cyan",
37: "white",
39: "default" # white.
}
#: An alias to :data:`~pyte.graphics.FG_ANSI` for compatibility.
FG = FG_ANSI
#: A mapping of non-standard ``aixterm`` foreground color codes to
#: color names. These are high intensity colors.
FG_AIXTERM = {
90: "brightblack",
91: "brightred",
92: "brightgreen",
93: "brightbrown",
94: "brightblue",
95: "brightmagenta",
96: "brightcyan",
97: "brightwhite"
}
#: A mapping of ANSI background color codes to color names.
#:
#: >>> BG_ANSI[40]
#: 'black'
#: >>> BG_ANSI[48]
#: 'default'
BG_ANSI = {
40: "black",
41: "red",
42: "green",
43: "brown",
44: "blue",
45: "magenta",
46: "cyan",
47: "white",
49: "default" # black.
}
#: An alias to :data:`~pyte.graphics.BG_ANSI` for compatibility.
BG = BG_ANSI
#: A mapping of non-standard ``aixterm`` background color codes to
#: color names. These are high intensity colors.
BG_AIXTERM = {
100: "brightblack",
101: "brightred",
102: "brightgreen",
103: "brightbrown",
104: "brightblue",
105: "bfightmagenta",
106: "brightcyan",
107: "brightwhite"
}
#: SGR code for foreground in 256 or True color mode.
FG_256 = 38
#: SGR code for background in 256 or True color mode.
BG_256 = 48
#: A table of 256 foreground or background colors.
# The following code is part of the Pygments project (BSD licensed).
_FG_BG_256 = [
(0x00, 0x00, 0x00), # 0
(0xcd, 0x00, 0x00), # 1
(0x00, 0xcd, 0x00), # 2
(0xcd, 0xcd, 0x00), # 3
(0x00, 0x00, 0xee), # 4
(0xcd, 0x00, 0xcd), # 5
(0x00, 0xcd, 0xcd), # 6
(0xe5, 0xe5, 0xe5), # 7
(0x7f, 0x7f, 0x7f), # 8
(0xff, 0x00, 0x00), # 9
(0x00, 0xff, 0x00), # 10
(0xff, 0xff, 0x00), # 11
(0x5c, 0x5c, 0xff), # 12
(0xff, 0x00, 0xff), # 13
(0x00, 0xff, 0xff), # 14
(0xff, 0xff, 0xff), # 15
]
# colors 16..231: the 6x6x6 color cube
valuerange = (0x00, 0x5f, 0x87, 0xaf, 0xd7, 0xff)
for i in range(216):
r = valuerange[(i // 36) % 6]
g = valuerange[(i // 6) % 6]
b = valuerange[i % 6]
_FG_BG_256.append((r, g, b))
# colors 232..255: grayscale
for i in range(24):
v = 8 + i * 10
_FG_BG_256.append((v, v, v))
FG_BG_256 = ["{0:02x}{1:02x}{2:02x}".format(r, g, b) for r, g, b in _FG_BG_256]
+61
View File
@@ -0,0 +1,61 @@
"""
pyte.modes
~~~~~~~~~~
This module defines terminal mode switches, used by
:class:`~pyte.screens.Screen`. There're two types of terminal modes:
* `non-private` which should be set with ``ESC [ N h``, where ``N``
is an integer, representing mode being set; and
* `private` which should be set with ``ESC [ ? N h``.
The latter are shifted 5 times to the right, to be easily
distinguishable from the former ones; for example `Origin Mode`
-- :data:`DECOM` is ``192`` not ``6``.
>>> DECOM
192
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
#: *Line Feed/New Line Mode*: When enabled, causes a received
#: :data:`~pyte.control.LF`, :data:`pyte.control.FF`, or
#: :data:`~pyte.control.VT` to move the cursor to the first column of
#: the next line.
LNM = 20
#: *Insert/Replace Mode*: When enabled, new display characters move
#: old display characters to the right. Characters moved past the
#: right margin are lost. Otherwise, new display characters replace
#: old display characters at the cursor position.
IRM = 4
# Private modes.
# ..............
#: *Text Cursor Enable Mode*: determines if the text cursor is
#: visible.
DECTCEM = 25 << 5
#: *Screen Mode*: toggles screen-wide reverse-video mode.
DECSCNM = 5 << 5
#: *Origin Mode*: allows cursor addressing relative to a user-defined
#: origin. This mode resets when the terminal is powered up or reset.
#: It does not affect the erase in display (ED) function.
DECOM = 6 << 5
#: *Auto Wrap Mode*: selects where received graphic characters appear
#: when the cursor is at the right margin.
DECAWM = 7 << 5
#: *Column Mode*: selects the number of columns per line (80 or 132)
#: on the screen.
DECCOLM = 3 << 5
+841
View File
@@ -0,0 +1,841 @@
"""
pyte.screens
~~~~~~~~~~~~
This module provides classes for terminal screens.
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
from __future__ import annotations
import copy
import json
import math
import os
import sys
import unicodedata
import warnings
from collections import deque, defaultdict
from functools import lru_cache
from typing import Any, Callable, DefaultDict, Dict, Generator, List, NamedTuple, Optional, Set, Sequence, TextIO, TypeVar
from wcwidth import wcwidth as _wcwidth # type: ignore[import]
from . import (
charsets as cs,
control as ctrl,
graphics as g,
modes as mo
)
from .streams import Stream
wcwidth: Callable[[str], int] = lru_cache(maxsize=4096)(_wcwidth)
KT = TypeVar("KT")
VT = TypeVar("VT")
# Clide diagnostic logging support
_debug_logger: Optional[Callable[[str], None]] = None
def set_debug_logger(logger: Optional[Callable[[str], None]]) -> None:
"""Set a debug logger function for diagnostic output."""
global _debug_logger
_debug_logger = logger
def _log_debug(message: str) -> None:
"""Log a debug message if debug logging is enabled."""
if _debug_logger is not None:
_debug_logger(message)
class Margins(NamedTuple):
"""A container for screen's scroll margins."""
top: int
bottom: int
class Savepoint(NamedTuple):
"""A container for savepoint, created on :data:`~pyte.escape.DECSC`."""
cursor: Cursor
g0_charset: str
g1_charset: str
charset: int
origin: bool
wrap: bool
class Char(NamedTuple):
"""A single styled on-screen character."""
data: str
fg: str = "default"
bg: str = "default"
bold: bool = False
italics: bool = False
underscore: bool = False
strikethrough: bool = False
reverse: bool = False
blink: bool = False
class Cursor:
"""Screen cursor."""
__slots__ = ("x", "y", "attrs", "hidden")
def __init__(self, x: int, y: int, attrs: Char = Char(" ")) -> None:
self.x = x
self.y = y
self.attrs = attrs
self.hidden = False
class StaticDefaultDict(Dict[KT, VT]):
"""A dict with a static default value."""
def __init__(self, default: VT) -> None:
self.default = default
def __missing__(self, key: KT) -> VT:
return self.default
_DEFAULT_MODE = set([mo.DECAWM, mo.DECTCEM])
class Screen:
"""A screen is an in-memory matrix of characters."""
@property
def default_char(self) -> Char:
"""An empty character with default foreground and background colors."""
reverse = mo.DECSCNM in self.mode
return Char(data=" ", fg="default", bg="default", reverse=reverse)
def __init__(self, columns: int, lines: int) -> None:
self.savepoints: List[Savepoint] = []
self.columns = columns
self.lines = lines
self.buffer: Dict[int, StaticDefaultDict[int, Char]] = defaultdict(lambda: StaticDefaultDict[int, Char](self.default_char))
self.dirty: Set[int] = set()
self.reset()
self.mode = _DEFAULT_MODE.copy()
self.margins: Optional[Margins] = None
def __repr__(self) -> str:
return ("{0}({1}, {2})".format(self.__class__.__name__,
self.columns, self.lines))
@property
def display(self) -> List[str]:
"""A list of screen lines as unicode strings."""
def render(line: StaticDefaultDict[int, Char]) -> Generator[str, None, None]:
is_wide_char = False
for x in range(self.columns):
if is_wide_char:
is_wide_char = False
continue
char = line[x].data
assert sum(map(wcwidth, char[1:])) == 0
is_wide_char = wcwidth(char[0]) == 2
yield char
return ["".join(render(self.buffer[y])) for y in range(self.lines)]
def reset(self) -> None:
"""Reset the terminal to its initial state."""
_log_debug("[SCREEN] reset()")
self.dirty.update(range(self.lines))
self.buffer.clear()
self.margins = None
self.mode = _DEFAULT_MODE.copy()
self.title = ""
self.icon_name = ""
self.charset = 0
self.g0_charset = cs.LAT1_MAP
self.g1_charset = cs.VT100_MAP
self.tabstops = set(range(8, self.columns, 8))
self.cursor = Cursor(0, 0)
self.cursor_position()
self.saved_columns: Optional[int] = None
def resize(self, lines: Optional[int] = None, columns: Optional[int] = None) -> None:
"""Resize the screen to the given size."""
lines = lines or self.lines
columns = columns or self.columns
if lines == self.lines and columns == self.columns:
return
_log_debug(f"[SCREEN] resize({lines}, {columns})")
self.dirty.update(range(lines))
if lines < self.lines:
self.save_cursor()
self.cursor_position(0, 0)
self.delete_lines(self.lines - lines)
self.restore_cursor()
if columns < self.columns:
for line in self.buffer.values():
for x in range(columns, self.columns):
line.pop(x, None)
self.lines, self.columns = lines, columns
self.set_margins()
def set_margins(self, top: Optional[int] = None, bottom: Optional[int] = None) -> None:
"""Select top and bottom margins for the scrolling region."""
if (top is None or top == 0) and bottom is None:
self.margins = None
return
margins = self.margins or Margins(0, self.lines - 1)
if top is None:
top = margins.top
else:
top = max(0, min(top - 1, self.lines - 1))
if bottom is None:
bottom = margins.bottom
else:
bottom = max(0, min(bottom - 1, self.lines - 1))
if bottom - top >= 1:
self.margins = Margins(top, bottom)
self.cursor_position()
def set_mode(self, *modes: int, **kwargs: Any) -> None:
"""Set (enable) a given list of modes."""
mode_list = list(modes)
if kwargs.get("private"):
mode_list = [mode << 5 for mode in modes]
if mo.DECSCNM in mode_list:
self.dirty.update(range(self.lines))
self.mode.update(mode_list)
if mo.DECCOLM in mode_list:
self.saved_columns = self.columns
self.resize(columns=132)
self.erase_in_display(2)
self.cursor_position()
if mo.DECOM in mode_list:
self.cursor_position()
if mo.DECSCNM in mode_list:
for line in self.buffer.values():
line.default = self.default_char
for x in line:
line[x] = line[x]._replace(reverse=True)
self.select_graphic_rendition(7)
if mo.DECTCEM in mode_list:
self.cursor.hidden = False
def reset_mode(self, *modes: int, **kwargs: Any) -> None:
"""Reset (disable) a given list of modes."""
mode_list = list(modes)
if kwargs.get("private"):
mode_list = [mode << 5 for mode in modes]
if mo.DECSCNM in mode_list:
self.dirty.update(range(self.lines))
self.mode.difference_update(mode_list)
if mo.DECCOLM in mode_list:
if self.columns == 132 and self.saved_columns is not None:
self.resize(columns=self.saved_columns)
self.saved_columns = None
self.erase_in_display(2)
self.cursor_position()
if mo.DECOM in mode_list:
self.cursor_position()
if mo.DECSCNM in mode_list:
for line in self.buffer.values():
line.default = self.default_char
for x in line:
line[x] = line[x]._replace(reverse=False)
self.select_graphic_rendition(27)
if mo.DECTCEM in mode_list:
self.cursor.hidden = True
def define_charset(self, code: str, mode: str) -> None:
"""Define G0 or G1 charset."""
if code in cs.MAPS:
if mode == "(":
self.g0_charset = cs.MAPS[code]
elif mode == ")":
self.g1_charset = cs.MAPS[code]
def shift_in(self) -> None:
"""Select G0 character set."""
self.charset = 0
def shift_out(self) -> None:
"""Select G1 character set."""
self.charset = 1
def draw(self, data: str) -> None:
"""Display decoded characters at the current cursor position."""
data = data.translate(
self.g1_charset if self.charset else self.g0_charset)
for char in data:
char_width = wcwidth(char)
# Clide: Log character drawing for debugging
if _debug_logger is not None and char_width > 0:
code = ord(char)
if code > 127 or code < 32:
_log_debug(f"[DRAW] char={char!r} code=U+{code:04X} width={char_width} pos=({self.cursor.x},{self.cursor.y})")
if self.cursor.x == self.columns:
if mo.DECAWM in self.mode:
self.dirty.add(self.cursor.y)
self.carriage_return()
self.linefeed()
elif char_width > 0:
self.cursor.x -= char_width
if mo.IRM in self.mode and char_width > 0:
self.insert_characters(char_width)
line = self.buffer[self.cursor.y]
if char_width == 1:
line[self.cursor.x] = self.cursor.attrs._replace(data=char)
elif char_width == 2:
line[self.cursor.x] = self.cursor.attrs._replace(data=char)
if self.cursor.x + 1 < self.columns:
line[self.cursor.x + 1] = self.cursor.attrs._replace(data="")
elif char_width == 0 and unicodedata.combining(char):
if self.cursor.x:
last = line[self.cursor.x - 1]
normalized = unicodedata.normalize("NFC", last.data + char)
line[self.cursor.x - 1] = last._replace(data=normalized)
elif self.cursor.y:
last = self.buffer[self.cursor.y - 1][self.columns - 1]
normalized = unicodedata.normalize("NFC", last.data + char)
self.buffer[self.cursor.y - 1][self.columns - 1] = last._replace(data=normalized)
else:
break
if char_width > 0:
self.cursor.x = min(self.cursor.x + char_width, self.columns)
self.dirty.add(self.cursor.y)
def set_title(self, param: str) -> None:
"""Set terminal title."""
self.title = param
def set_icon_name(self, param: str) -> None:
"""Set icon name."""
self.icon_name = param
def carriage_return(self) -> None:
"""Move the cursor to the beginning of the current line."""
self.cursor.x = 0
def index(self) -> None:
"""Move the cursor down one line in the same column."""
top, bottom = self.margins or Margins(0, self.lines - 1)
if self.cursor.y == bottom:
self.dirty.update(range(self.lines))
for y in range(top, bottom):
self.buffer[y] = self.buffer[y + 1]
self.buffer.pop(bottom, None)
else:
self.cursor_down()
def reverse_index(self) -> None:
"""Move the cursor up one line in the same column."""
top, bottom = self.margins or Margins(0, self.lines - 1)
if self.cursor.y == top:
self.dirty.update(range(self.lines))
for y in range(bottom, top, -1):
self.buffer[y] = self.buffer[y - 1]
self.buffer.pop(top, None)
else:
self.cursor_up()
def linefeed(self) -> None:
"""Perform an index and, if LNM is set, a carriage return."""
self.index()
if mo.LNM in self.mode:
self.carriage_return()
def tab(self) -> None:
"""Move to the next tab space."""
for stop in sorted(self.tabstops):
if self.cursor.x < stop:
column = stop
break
else:
column = self.columns - 1
self.cursor.x = column
def backspace(self) -> None:
"""Move cursor to the left one."""
self.cursor_back()
def save_cursor(self) -> None:
"""Push the current cursor position onto the stack."""
self.savepoints.append(Savepoint(copy.copy(self.cursor),
self.g0_charset,
self.g1_charset,
self.charset,
mo.DECOM in self.mode,
mo.DECAWM in self.mode))
def restore_cursor(self) -> None:
"""Set the current cursor position to whatever cursor is on top of the stack."""
if self.savepoints:
savepoint = self.savepoints.pop()
self.g0_charset = savepoint.g0_charset
self.g1_charset = savepoint.g1_charset
self.charset = savepoint.charset
if savepoint.origin:
self.set_mode(mo.DECOM)
if savepoint.wrap:
self.set_mode(mo.DECAWM)
self.cursor = savepoint.cursor
self.ensure_hbounds()
self.ensure_vbounds(use_margins=True)
else:
self.reset_mode(mo.DECOM)
self.cursor_position()
def insert_lines(self, count: Optional[int] = None) -> None:
"""Insert the indicated # of lines at line with cursor."""
count = count or 1
top, bottom = self.margins or Margins(0, self.lines - 1)
if top <= self.cursor.y <= bottom:
self.dirty.update(range(self.cursor.y, self.lines))
for y in range(bottom, self.cursor.y - 1, -1):
if y + count <= bottom and y in self.buffer:
self.buffer[y + count] = self.buffer[y]
self.buffer.pop(y, None)
self.carriage_return()
def delete_lines(self, count: Optional[int] = None) -> None:
"""Delete the indicated # of lines."""
count = count or 1
top, bottom = self.margins or Margins(0, self.lines - 1)
if top <= self.cursor.y <= bottom:
self.dirty.update(range(self.cursor.y, self.lines))
for y in range(self.cursor.y, bottom + 1):
if y + count <= bottom:
if y + count in self.buffer:
self.buffer[y] = self.buffer.pop(y + count)
else:
self.buffer.pop(y, None)
self.carriage_return()
def insert_characters(self, count: Optional[int] = None) -> None:
"""Insert the indicated # of blank characters at the cursor position."""
self.dirty.add(self.cursor.y)
count = count or 1
line = self.buffer[self.cursor.y]
for x in range(self.columns, self.cursor.x - 1, -1):
if x + count <= self.columns:
line[x + count] = line[x]
line.pop(x, None)
def delete_characters(self, count: Optional[int] = None) -> None:
"""Delete the indicated # of characters."""
self.dirty.add(self.cursor.y)
count = count or 1
line = self.buffer[self.cursor.y]
for x in range(self.cursor.x, self.columns):
if x + count <= self.columns:
line[x] = line.pop(x + count, self.default_char)
else:
line.pop(x, None)
def erase_characters(self, count: Optional[int] = None) -> None:
"""Erase the indicated # of characters."""
self.dirty.add(self.cursor.y)
count = count or 1
line = self.buffer[self.cursor.y]
for x in range(self.cursor.x, min(self.cursor.x + count, self.columns)):
line[x] = self.cursor.attrs
def erase_in_line(self, how: int = 0, private: bool = False) -> None:
"""Erase a line in a specific way."""
self.dirty.add(self.cursor.y)
if how == 0:
interval = range(self.cursor.x, self.columns)
elif how == 1:
interval = range(self.cursor.x + 1)
elif how == 2:
interval = range(self.columns)
line = self.buffer[self.cursor.y]
for x in interval:
line[x] = self.cursor.attrs
def erase_in_display(self, how: int = 0, *args: Any, **kwargs: Any) -> None:
"""Erases display in a specific way."""
_log_debug(f"[SCREEN] erase_in_display(how={how})")
if how == 0:
interval = range(self.cursor.y + 1, self.lines)
elif how == 1:
interval = range(self.cursor.y)
elif how == 2 or how == 3:
interval = range(self.lines)
self.dirty.update(interval)
for y in interval:
line = self.buffer[y]
for x in line:
line[x] = self.cursor.attrs
if how == 0 or how == 1:
self.erase_in_line(how)
def set_tab_stop(self) -> None:
"""Set a horizontal tab stop at cursor position."""
self.tabstops.add(self.cursor.x)
def clear_tab_stop(self, how: int = 0) -> None:
"""Clear a horizontal tab stop."""
if how == 0:
self.tabstops.discard(self.cursor.x)
elif how == 3:
self.tabstops = set()
def ensure_hbounds(self) -> None:
"""Ensure the cursor is within horizontal screen bounds."""
self.cursor.x = min(max(0, self.cursor.x), self.columns - 1)
def ensure_vbounds(self, use_margins: Optional[bool] = None) -> None:
"""Ensure the cursor is within vertical screen bounds."""
if (use_margins or mo.DECOM in self.mode) and self.margins is not None:
top, bottom = self.margins
else:
top, bottom = 0, self.lines - 1
self.cursor.y = min(max(top, self.cursor.y), bottom)
def cursor_up(self, count: Optional[int] = None) -> None:
"""Move cursor up the indicated # of lines."""
top, _bottom = self.margins or Margins(0, self.lines - 1)
self.cursor.y = max(self.cursor.y - (count or 1), top)
def cursor_up1(self, count: Optional[int] = None) -> None:
"""Move cursor up the indicated # of lines to column 1."""
self.cursor_up(count)
self.carriage_return()
def cursor_down(self, count: Optional[int] = None) -> None:
"""Move cursor down the indicated # of lines."""
_top, bottom = self.margins or Margins(0, self.lines - 1)
self.cursor.y = min(self.cursor.y + (count or 1), bottom)
def cursor_down1(self, count: Optional[int] = None) -> None:
"""Move cursor down the indicated # of lines to column 1."""
self.cursor_down(count)
self.carriage_return()
def cursor_back(self, count: Optional[int] = None) -> None:
"""Move cursor left the indicated # of columns."""
if self.cursor.x == self.columns:
self.cursor.x -= 1
self.cursor.x -= count or 1
self.ensure_hbounds()
def cursor_forward(self, count: Optional[int] = None) -> None:
"""Move cursor right the indicated # of columns."""
self.cursor.x += count or 1
self.ensure_hbounds()
def cursor_position(self, line: Optional[int] = None, column: Optional[int] = None) -> None:
"""Set the cursor to a specific line and column."""
column = (column or 1) - 1
line = (line or 1) - 1
if self.margins is not None and mo.DECOM in self.mode:
line += self.margins.top
if not self.margins.top <= line <= self.margins.bottom:
return
self.cursor.x = column
self.cursor.y = line
self.ensure_hbounds()
self.ensure_vbounds()
def cursor_to_column(self, column: Optional[int] = None) -> None:
"""Move cursor to a specific column in the current line."""
self.cursor.x = (column or 1) - 1
self.ensure_hbounds()
def cursor_to_line(self, line: Optional[int] = None) -> None:
"""Move cursor to a specific line in the current column."""
self.cursor.y = (line or 1) - 1
if mo.DECOM in self.mode:
assert self.margins is not None
self.cursor.y += self.margins.top
self.ensure_vbounds()
def bell(self, *args: Any) -> None:
"""Bell stub."""
pass
def alignment_display(self) -> None:
"""Fills screen with uppercase E's for screen focus and alignment."""
self.dirty.update(range(self.lines))
for y in range(self.lines):
for x in range(self.columns):
self.buffer[y][x] = self.buffer[y][x]._replace(data="E")
def select_graphic_rendition(self, *attrs: int) -> None:
"""Set display attributes."""
replace = {}
if not attrs or attrs == (0, ):
self.cursor.attrs = self.default_char
return
attrs_list = list(reversed(attrs))
while attrs_list:
attr = attrs_list.pop()
if attr == 0:
replace.update(self.default_char._asdict())
elif attr in g.FG_ANSI:
replace["fg"] = g.FG_ANSI[attr]
elif attr in g.BG:
replace["bg"] = g.BG_ANSI[attr]
elif attr in g.TEXT:
attr_str = g.TEXT[attr]
replace[attr_str[1:]] = attr_str.startswith("+")
elif attr in g.FG_AIXTERM:
replace.update(fg=g.FG_AIXTERM[attr])
elif attr in g.BG_AIXTERM:
replace.update(bg=g.BG_AIXTERM[attr])
elif attr in (g.FG_256, g.BG_256):
key = "fg" if attr == g.FG_256 else "bg"
try:
n = attrs_list.pop()
if n == 5:
m = attrs_list.pop()
replace[key] = g.FG_BG_256[m]
elif n == 2:
replace[key] = "{0:02x}{1:02x}{2:02x}".format(
attrs_list.pop(), attrs_list.pop(), attrs_list.pop())
except IndexError:
pass
self.cursor.attrs = self.cursor.attrs._replace(**replace)
def report_device_attributes(self, mode: int = 0, **kwargs: bool) -> None:
"""Report terminal identity."""
if mode == 0 and not kwargs.get("private"):
self.write_process_input(ctrl.CSI + "?6c")
def report_device_status(self, mode: int) -> None:
"""Report terminal status or cursor position."""
if mode == 5:
self.write_process_input(ctrl.CSI + "0n")
elif mode == 6:
x = self.cursor.x + 1
y = self.cursor.y + 1
if mo.DECOM in self.mode:
assert self.margins is not None
y -= self.margins.top
self.write_process_input(ctrl.CSI + "{0};{1}R".format(y, x))
def write_process_input(self, data: str) -> None:
"""Write data to the process running inside the terminal."""
pass
def debug(self, *args: Any, **kwargs: Any) -> None:
"""Endpoint for unrecognized escape sequences."""
if _debug_logger is not None:
_log_debug(f"[DEBUG] unrecognized: args={args} kwargs={kwargs}")
class DiffScreen(Screen):
"""A screen subclass, which maintains a set of dirty lines. Deprecated."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
warnings.warn(
"The functionality of ``DiffScreen` has been merged into "
"``Screen`` and will be removed in 0.8.0.", DeprecationWarning)
super(DiffScreen, self).__init__(*args, **kwargs)
class History(NamedTuple):
top: deque[StaticDefaultDict[int, Char]]
bottom: deque[StaticDefaultDict[int, Char]]
ratio: float
size: int
position: int
class HistoryScreen(Screen):
"""A Screen subclass, which keeps track of screen history."""
_wrapped = set(Stream.events)
_wrapped.update(["next_page", "prev_page"])
def __init__(self, columns: int, lines: int, history: int = 100, ratio: float = .5) -> None:
self.history = History(deque(maxlen=history),
deque(maxlen=history),
float(ratio),
history,
history)
super(HistoryScreen, self).__init__(columns, lines)
def _make_wrapper(self, event: str, handler: Callable[..., Any]) -> Callable[..., Any]:
def inner(*args: Any, **kwargs: Any) -> Any:
self.before_event(event)
result = handler(*args, **kwargs)
self.after_event(event)
return result
return inner
def __getattribute__(self, attr: str) -> Callable[..., Any]:
value = super(HistoryScreen, self).__getattribute__(attr)
if attr in HistoryScreen._wrapped:
return HistoryScreen._make_wrapper(self, attr, value)
else:
return value # type: ignore[no-any-return]
def before_event(self, event: str) -> None:
"""Ensure a screen is at the bottom of the history buffer."""
if event not in ["prev_page", "next_page"]:
while self.history.position < self.history.size:
self.next_page()
def after_event(self, event: str) -> None:
"""Ensure all lines on a screen have proper width."""
if event in ["prev_page", "next_page"]:
for line in self.buffer.values():
for x in line:
if x > self.columns:
line.pop(x)
self.cursor.hidden = not (
self.history.position == self.history.size and
mo.DECTCEM in self.mode
)
def _reset_history(self) -> None:
self.history.top.clear()
self.history.bottom.clear()
self.history = self.history._replace(position=self.history.size)
def reset(self) -> None:
"""Overloaded to reset screen history state."""
super(HistoryScreen, self).reset()
self._reset_history()
def erase_in_display(self, how: int = 0, *args: Any, **kwargs: Any) -> None:
"""Overloaded to reset history state."""
super(HistoryScreen, self).erase_in_display(how, *args, **kwargs)
if how == 3:
self._reset_history()
def index(self) -> None:
"""Overloaded to update top history with the removed lines."""
top, bottom = self.margins or Margins(0, self.lines - 1)
if self.cursor.y == bottom:
self.history.top.append(self.buffer[top])
super(HistoryScreen, self).index()
def reverse_index(self) -> None:
"""Overloaded to update bottom history with the removed lines."""
top, bottom = self.margins or Margins(0, self.lines - 1)
if self.cursor.y == top:
self.history.bottom.append(self.buffer[bottom])
super(HistoryScreen, self).reverse_index()
def prev_page(self) -> None:
"""Move the screen page up through the history buffer."""
if self.history.position > self.lines and self.history.top:
mid = min(len(self.history.top),
int(math.ceil(self.lines * self.history.ratio)))
self.history.bottom.extendleft(
self.buffer[y]
for y in range(self.lines - 1, self.lines - mid - 1, -1))
self.history = self.history._replace(position=self.history.position - mid)
for y in range(self.lines - 1, mid - 1, -1):
self.buffer[y] = self.buffer[y - mid]
for y in range(mid - 1, -1, -1):
self.buffer[y] = self.history.top.pop()
self.dirty = set(range(self.lines))
def next_page(self) -> None:
"""Move the screen page down through the history buffer."""
if self.history.position < self.history.size and self.history.bottom:
mid = min(len(self.history.bottom),
int(math.ceil(self.lines * self.history.ratio)))
self.history.top.extend(self.buffer[y] for y in range(mid))
self.history = self.history._replace(position=self.history.position + mid)
for y in range(self.lines - mid):
self.buffer[y] = self.buffer[y + mid]
for y in range(self.lines - mid, self.lines):
self.buffer[y] = self.history.bottom.popleft()
self.dirty = set(range(self.lines))
class DebugEvent(NamedTuple):
"""Event dispatched to DebugScreen."""
name: str
args: Any
kwargs: Any
@staticmethod
def from_string(line: str) -> DebugEvent:
return DebugEvent(*json.loads(line))
def __str__(self) -> str:
return json.dumps(self)
def __call__(self, screen: Screen) -> Any:
"""Execute this event on a given screen."""
return getattr(screen, self.name)(*self.args, **self.kwargs)
class DebugScreen:
"""A screen which dumps a subset of the received events to a file."""
def __init__(self, to: TextIO = sys.stderr, only: Sequence[str] = ()) -> None:
self.to = to
self.only = only
def only_wrapper(self, attr: str) -> Callable[..., None]:
def wrapper(*args: Any, **kwargs: Any) -> None:
self.to.write(str(DebugEvent(attr, args, kwargs)))
self.to.write(str(os.linesep))
return wrapper
def __getattribute__(self, attr: str) -> Callable[..., None]:
if attr not in Stream.events:
return super(DebugScreen, self).__getattribute__(attr) # type: ignore[no-any-return]
elif not self.only or attr in self.only:
return self.only_wrapper(attr)
else:
return lambda *args, **kwargs: None
+506
View File
@@ -0,0 +1,506 @@
"""
pyte.streams
~~~~~~~~~~~~
This module provides three stream implementations with different
features; for starters, here's a quick example of how streams are
typically used:
>>> import pyte
>>> screen = pyte.Screen(80, 24)
>>> stream = pyte.Stream(screen)
>>> stream.feed("\x1b[5B") # Move the cursor down 5 rows.
>>> screen.cursor.y
5
:copyright: (c) 2011-2012 by Selectel.
:copyright: (c) 2012-2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
Vendored for Clide with modifications for diagnostic logging.
"""
from __future__ import annotations
import codecs
import itertools
import re
import warnings
from collections import defaultdict
from collections.abc import Mapping
from typing import Any, Callable, Dict, Generator, Optional, TYPE_CHECKING
from . import control as ctrl, escape as esc
if TYPE_CHECKING:
from .screens import Screen
# Clide diagnostic logging support
_debug_logger: Optional[Callable[[str], None]] = None
_event_callback: Optional[Callable[[str], None]] = None
def set_debug_logger(logger: Optional[Callable[[str], None]]) -> None:
"""Set a debug logger function for diagnostic output.
Args:
logger: A callable that accepts a string message, or None to disable.
"""
global _debug_logger
_debug_logger = logger
def set_event_callback(callback: Optional[Callable[[str], None]]) -> None:
"""Set an event callback for raw terminal data.
This callback is invoked with the raw data before parsing.
Useful for Claude Code event detection.
Args:
callback: A callable that accepts raw terminal data, or None to disable.
"""
global _event_callback
_event_callback = callback
def _log_debug(message: str) -> None:
"""Log a debug message if debug logging is enabled."""
if _debug_logger is not None:
_debug_logger(message)
ParserGenerator = Generator[Optional[bool], str, None]
class Stream:
"""A stream is a state machine that parses a stream of bytes and
dispatches events based on what it sees.
:param pyte.screens.Screen screen: a screen to dispatch events to.
:param bool strict: check if a given screen implements all required
events.
.. note::
Stream only accepts text as input, but if for some reason
you need to feed it with bytes, consider using
:class:`~pyte.streams.ByteStream` instead.
.. versionchanged 0.6.0::
For performance reasons the binding between stream events and
screen methods was made static. As a result, the stream **will
not** dispatch events to methods added to screen **after** the
stream was created.
.. seealso::
`man console_codes <http://linux.die.net/man/4/console_codes>`_
For details on console codes listed bellow in :attr:`basic`,
:attr:`escape`, :attr:`csi`, :attr:`sharp`.
"""
#: Control sequences, which don't require any arguments.
basic = {
ctrl.BEL: "bell",
ctrl.BS: "backspace",
ctrl.HT: "tab",
ctrl.LF: "linefeed",
ctrl.VT: "linefeed",
ctrl.FF: "linefeed",
ctrl.CR: "carriage_return",
ctrl.SO: "shift_out",
ctrl.SI: "shift_in",
}
#: non-CSI escape sequences.
escape = {
esc.RIS: "reset",
esc.IND: "index",
esc.NEL: "linefeed",
esc.RI: "reverse_index",
esc.HTS: "set_tab_stop",
esc.DECSC: "save_cursor",
esc.DECRC: "restore_cursor",
}
#: "sharp" escape sequences -- ``ESC # <N>``.
sharp = {
esc.DECALN: "alignment_display",
}
#: CSI escape sequences -- ``CSI P1;P2;...;Pn <fn>``.
csi = {
esc.ICH: "insert_characters",
esc.CUU: "cursor_up",
esc.CUD: "cursor_down",
esc.CUF: "cursor_forward",
esc.CUB: "cursor_back",
esc.CNL: "cursor_down1",
esc.CPL: "cursor_up1",
esc.CHA: "cursor_to_column",
esc.CUP: "cursor_position",
esc.ED: "erase_in_display",
esc.EL: "erase_in_line",
esc.IL: "insert_lines",
esc.DL: "delete_lines",
esc.DCH: "delete_characters",
esc.ECH: "erase_characters",
esc.HPR: "cursor_forward",
esc.DA: "report_device_attributes",
esc.VPA: "cursor_to_line",
esc.VPR: "cursor_down",
esc.HVP: "cursor_position",
esc.TBC: "clear_tab_stop",
esc.SM: "set_mode",
esc.RM: "reset_mode",
esc.SGR: "select_graphic_rendition",
esc.DSR: "report_device_status",
esc.DECSTBM: "set_margins",
esc.HPA: "cursor_to_column"
}
#: A set of all events dispatched by the stream.
events = frozenset(itertools.chain(
basic.values(), escape.values(), sharp.values(), csi.values(),
["define_charset"],
["set_icon_name", "set_title"], # OSC.
["draw", "debug"]))
#: A regular expression pattern matching everything what can be
#: considered plain text.
_special = set([ctrl.ESC, ctrl.CSI_C1, ctrl.NUL, ctrl.DEL, ctrl.OSC_C1])
_special.update(basic)
_text_pattern = re.compile(
"[^" + "".join(map(re.escape, _special)) + "]+")
del _special
def __init__(self, screen: Optional[Screen] = None, strict: bool = True) -> None:
self.listener: Optional[Screen] = None
self.strict = strict
self.use_utf8: bool = True
self._taking_plain_text: Optional[bool] = None
if screen is not None:
self.attach(screen)
def attach(self, screen: Screen) -> None:
"""Adds a given screen to the listener queue.
:param pyte.screens.Screen screen: a screen to attach to.
"""
if self.listener is not None:
warnings.warn("As of version 0.6.0 the listener queue is "
"restricted to a single element. Existing "
"listener {0} will be replaced."
.format(self.listener), DeprecationWarning)
if self.strict:
for event in self.events:
if not hasattr(screen, event):
raise TypeError("{0} is missing {1}".format(screen, event))
self.listener = screen
self._parser: Optional[ParserGenerator] = None
self._initialize_parser()
def detach(self, screen: Screen) -> None:
"""Remove a given screen from the listener queue and fails
silently if it's not attached.
:param pyte.screens.Screen screen: a screen to detach.
"""
if screen is self.listener:
self.listener = None
def feed(self, data: str) -> None:
"""Consume some data and advances the state as necessary.
:param str data: a blob of data to feed from.
"""
# Clide: Invoke event callback for raw data
if _event_callback is not None:
try:
_event_callback(data)
except Exception:
pass # Don't let callback errors affect parsing
# Clide: Debug log incoming data
if _debug_logger is not None:
# Log escape sequences in a readable format
escaped = data.encode('unicode_escape').decode('ascii')
if len(escaped) > 200:
escaped = escaped[:200] + "..."
_log_debug(f"[STREAM] feed: {escaped}")
send = self._send_to_parser
if self.listener is None:
raise RuntimeError("Listener is not set")
draw = self.listener.draw
match_text = self._text_pattern.match
taking_plain_text = self._taking_plain_text
length = len(data)
offset = 0
while offset < length:
if taking_plain_text:
match = match_text(data, offset)
if match:
start, offset = match.span()
draw(data[start:offset])
else:
taking_plain_text = False
else:
taking_plain_text = send(data[offset:offset + 1])
offset += 1
self._taking_plain_text = taking_plain_text
def _send_to_parser(self, data: str) -> Optional[bool]:
try:
assert self._parser is not None
return self._parser.send(data)
except Exception:
# Reset the parser state to make sure it is usable even
# after receiving an exception. See PR #101 for details.
self._initialize_parser()
raise
def _initialize_parser(self) -> None:
self._parser = self._parser_fsm()
self._taking_plain_text = next(self._parser)
def _parser_fsm(self) -> ParserGenerator:
"""An FSM implemented as a coroutine.
This generator is not the most beautiful, but it is as performant
as possible. When a process generates a lot of output, then this
will be the bottleneck, because it processes just one character
at a time.
Don't change anything without profiling first.
"""
basic = self.basic
assert self.listener is not None
listener = self.listener
draw = listener.draw
debug = listener.debug
ESC, CSI_C1 = ctrl.ESC, ctrl.CSI_C1
OSC_C1 = ctrl.OSC_C1
SP_OR_GT = ctrl.SP + ">"
NUL_OR_DEL = ctrl.NUL + ctrl.DEL
CAN_OR_SUB = ctrl.CAN + ctrl.SUB
ALLOWED_IN_CSI = "".join([ctrl.BEL, ctrl.BS, ctrl.HT, ctrl.LF,
ctrl.VT, ctrl.FF, ctrl.CR])
OSC_TERMINATORS = set([ctrl.ST_C0, ctrl.ST_C1, ctrl.BEL])
def create_dispatcher(mapping: Mapping[str, str]) -> Dict[str, Callable[..., None]]:
d = defaultdict(lambda: debug, dict(
(event, getattr(listener, attr))
for event, attr in mapping.items()))
# Clide: Wrap dispatchers with debug logging
if _debug_logger is not None:
original_d = dict(d)
for event, attr in mapping.items():
original_handler = original_d.get(event, debug)
def make_wrapper(e: str, a: str, h: Callable[..., None]) -> Callable[..., None]:
def wrapper(*args: Any, **kwargs: Any) -> None:
_log_debug(f"[DISPATCH] {a}({args}, {kwargs})")
return h(*args, **kwargs)
return wrapper
d[event] = make_wrapper(event, attr, original_handler)
return d
basic_dispatch = create_dispatcher(basic)
sharp_dispatch = create_dispatcher(self.sharp)
escape_dispatch = create_dispatcher(self.escape)
csi_dispatch = create_dispatcher(self.csi)
while True:
# ``True`` tells ``Screen.feed`` that it is allowed to send
# chunks of plain text directly to the listener, instead
# of this generator.
char = yield True
if char == ESC:
# Most non-VT52 commands start with a left-bracket after the
# escape and then a stream of parameters and a command; with
# a single notable exception -- :data:`escape.DECOM` sequence,
# which starts with a sharp.
#
# .. versionchanged:: 0.4.10
#
# For compatibility with Linux terminal stream also
# recognizes ``ESC % C`` sequences for selecting control
# character set. However, in the current version these
# are noop.
char = yield None
if char == "[":
char = CSI_C1 # Go to CSI.
elif char == "]":
char = OSC_C1 # Go to OSC.
else:
if char == "#":
sharp_dispatch[(yield None)]()
elif char == "%":
self.select_other_charset((yield None))
elif char in "()":
code = yield None
if self.use_utf8:
continue
# See http://www.cl.cam.ac.uk/~mgk25/unicode.html#term
# for the why on the UTF-8 restriction.
listener.define_charset(code, mode=char)
else:
escape_dispatch[char]()
continue # Don't go to CSI.
if char in basic:
# Ignore shifts in UTF-8 mode. See
# http://www.cl.cam.ac.uk/~mgk25/unicode.html#term for
# the why on UTF-8 restriction.
if (char == ctrl.SI or char == ctrl.SO) and self.use_utf8:
continue
basic_dispatch[char]()
elif char == CSI_C1:
# All parameters are unsigned, positive decimal integers, with
# the most significant digit sent first. Any parameter greater
# than 9999 is set to 9999. If you do not specify a value, a 0
# value is assumed.
#
# .. seealso::
#
# `VT102 User Guide <http://vt100.net/docs/vt102-ug/>`_
# For details on the formatting of escape arguments.
#
# `VT220 Programmer Ref. <http://vt100.net/docs/vt220-rm/>`_
# For details on the characters valid for use as
# arguments.
params = []
current = ""
private = False
while True:
char = yield None
if char == "?":
private = True
elif char in ALLOWED_IN_CSI:
basic_dispatch[char]()
elif char in SP_OR_GT:
pass # Secondary DA is not supported atm.
elif char in CAN_OR_SUB:
# If CAN or SUB is received during a sequence, the
# current sequence is aborted; terminal displays
# the substitute character, followed by characters
# in the sequence received after CAN or SUB.
draw(char)
break
elif char.isdigit():
current += char
elif char == "$":
# XTerm-specific ESC]...$[a-z] sequences are not
# currently supported.
yield None
break
else:
params.append(min(int(current or 0), 9999))
if char == ";":
current = ""
else:
# Clide: Log CSI sequence
if _debug_logger is not None:
_log_debug(f"[CSI] char={char!r} params={params} private={private}")
if private:
csi_dispatch[char](*params, private=True)
else:
csi_dispatch[char](*params)
break # CSI is finished.
elif char == OSC_C1:
code = yield None
if code == "R":
continue # Reset palette. Not implemented.
elif code == "P":
continue # Set palette. Not implemented.
param = ""
while True:
char = yield None
if char == ESC:
char += yield None
if char in OSC_TERMINATORS:
break
else:
param += char
param = param[1:] # Drop the ;.
# Clide: Log OSC sequence
if _debug_logger is not None:
_log_debug(f"[OSC] code={code!r} param={param!r}")
if code in "01":
listener.set_icon_name(param)
if code in "02":
listener.set_title(param)
elif char not in NUL_OR_DEL:
draw(char)
def select_other_charset(self, code: str) -> None:
"""Select other (non G0 or G1) charset.
:param str code: character set code, should be a character from
``"@G8"``, otherwise ignored.
.. note:: We currently follow ``"linux"`` and only use this
command to switch from ISO-8859-1 to UTF-8 and back.
.. versionadded:: 0.6.0
.. seealso::
`Standard ECMA-35, Section 15.4 \
<http://ecma-international.org/publications/standards/Ecma-035.htm>`_
for a description of VTXXX character set machinery.
"""
# A noop since all input is Unicode-only.
class ByteStream(Stream):
"""A stream which takes bytes as input.
Bytes are decoded to text using either UTF-8 (default) or the encoding
selected via :meth:`~pyte.Stream.select_other_charset`.
.. attribute:: use_utf8
Assume the input to :meth:`~pyte.streams.ByteStream.feed` is encoded
using UTF-8. Defaults to ``True``.
"""
def __init__(self, *args: Any, **kwargs: Any):
super(ByteStream, self).__init__(*args, **kwargs)
self.utf8_decoder = codecs.getincrementaldecoder("utf-8")("replace")
def feed(self, data: bytes) -> None: # type: ignore[override]
if self.use_utf8:
data_str = self.utf8_decoder.decode(data)
else:
data_str = "".join(map(chr, data))
super(ByteStream, self).feed(data_str)
def select_other_charset(self, code: str) -> None:
if code == "@":
self.use_utf8 = False
self.utf8_decoder.reset()
elif code in "G8":
self.use_utf8 = True
+322
View File
@@ -0,0 +1,322 @@
"""Action bar widget for contextual toolbar buttons."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable
from textual.app import ComposeResult
from textual.containers import Horizontal
from textual.message import Message
from textual.reactive import reactive
from textual.widget import Widget
from textual.widgets import Button, Static
@dataclass
class ActionButton:
"""Definition for an action button.
Attributes:
id: Unique identifier for the button
icon: Unicode icon to display
tooltip: Hover text / description
callback: Function to call when clicked (or None for message-based)
visible: Whether button is currently visible
enabled: Whether button is currently enabled
"""
id: str
icon: str
tooltip: str
callback: Callable[[], None] | None = None
visible: bool = True
enabled: bool = True
class ActionBarButton(Static):
"""A compact action bar button using Static for cleaner rendering."""
DEFAULT_CSS = """
ActionBarButton {
width: auto;
height: 1;
padding: 0 1;
margin: 0;
color: $text-muted;
}
ActionBarButton:hover {
background: $surface-lighten-1;
color: $text;
}
ActionBarButton.-active {
color: $primary;
}
ActionBarButton.-disabled {
color: $text-disabled;
}
"""
can_focus = True
def __init__(
self,
icon: str,
tooltip: str,
action_id: str,
disabled: bool = False,
**kwargs,
) -> None:
super().__init__(icon, **kwargs)
self.tooltip = tooltip
self.action_id = action_id
self._disabled = disabled
if disabled:
self.add_class("-disabled")
@property
def disabled(self) -> bool:
return self._disabled
@disabled.setter
def disabled(self, value: bool) -> None:
self._disabled = value
if value:
self.add_class("-disabled")
else:
self.remove_class("-disabled")
def on_click(self, event) -> None:
"""Handle click events."""
if not self._disabled:
# Post a button pressed message
self.post_message(Button.Pressed(self))
class ActionBar(Horizontal):
"""Contextual action bar for workspace panels.
Displays action buttons that can be dynamically added/removed
based on the active context (editor, diff, terminal, etc.).
Example:
action_bar = ActionBar()
action_bar.register_button(ActionButton(
id="save",
icon="💾",
tooltip="Save file",
callback=self.save_file,
))
"""
DEFAULT_CSS = """
ActionBar {
width: auto;
height: auto;
padding: 0;
}
ActionBar .action-separator {
width: 1;
height: 1;
margin: 0;
color: $text-muted;
}
"""
class ButtonPressed(Message):
"""Emitted when an action button is pressed."""
def __init__(self, button_id: str) -> None:
self.button_id = button_id
super().__init__()
# Track maximized state
maximized: reactive[bool] = reactive(False)
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._buttons: dict[str, ActionButton] = {}
self._button_order: list[str] = []
def compose(self) -> ComposeResult:
"""Compose the action bar - buttons added dynamically."""
# Initially empty, buttons added via register_button
yield from []
def register_button(
self,
button: ActionButton,
*,
position: int | None = None,
) -> None:
"""Register an action button.
Args:
button: The button definition
position: Optional position in the bar (default: end)
"""
self._buttons[button.id] = button
if position is not None:
self._button_order.insert(position, button.id)
else:
self._button_order.append(button.id)
# Create and mount the button widget
btn_widget = ActionBarButton(
icon=button.icon,
tooltip=button.tooltip,
action_id=button.id,
id=f"action-{button.id}",
disabled=not button.enabled,
)
if not button.visible:
btn_widget.display = False
self.mount(btn_widget)
def unregister_button(self, button_id: str) -> None:
"""Remove an action button.
Args:
button_id: ID of the button to remove
"""
if button_id in self._buttons:
del self._buttons[button_id]
self._button_order.remove(button_id)
try:
btn = self.query_one(f"#action-{button_id}", ActionBarButton)
btn.remove()
except Exception:
pass
def set_button_visible(self, button_id: str, visible: bool) -> None:
"""Show or hide a button.
Args:
button_id: ID of the button
visible: Whether to show the button
"""
if button_id in self._buttons:
self._buttons[button_id].visible = visible
try:
btn = self.query_one(f"#action-{button_id}", ActionBarButton)
btn.display = visible
except Exception:
pass
def set_button_enabled(self, button_id: str, enabled: bool) -> None:
"""Enable or disable a button.
Args:
button_id: ID of the button
enabled: Whether to enable the button
"""
if button_id in self._buttons:
self._buttons[button_id].enabled = enabled
try:
btn = self.query_one(f"#action-{button_id}", ActionBarButton)
btn.disabled = not enabled
except Exception:
pass
def set_button_active(self, button_id: str, active: bool) -> None:
"""Set a button's active state (visual highlight).
Args:
button_id: ID of the button
active: Whether button should appear active
"""
try:
btn = self.query_one(f"#action-{button_id}", ActionBarButton)
if active:
btn.add_class("-active")
else:
btn.remove_class("-active")
except Exception:
pass
def update_button_icon(self, button_id: str, icon: str) -> None:
"""Update a button's icon.
Args:
button_id: ID of the button
icon: New icon to display
"""
if button_id in self._buttons:
self._buttons[button_id].icon = icon
try:
btn = self.query_one(f"#action-{button_id}", ActionBarButton)
btn.label = icon
except Exception:
pass
def add_separator(self) -> None:
"""Add a visual separator."""
sep = Static("", classes="action-separator")
self.mount(sep)
def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button press."""
if isinstance(event.button, ActionBarButton):
button_id = event.button.action_id
# Call the callback if defined
if button_id in self._buttons:
button = self._buttons[button_id]
if button.callback:
button.callback()
# Also emit a message for flexible handling
self.post_message(self.ButtonPressed(button_id))
def clear(self) -> None:
"""Remove all buttons."""
for btn_id in list(self._buttons.keys()):
self.unregister_button(btn_id)
# Also remove any separators
for sep in self.query(".action-separator"):
sep.remove()
# Standard action button definitions for common operations
STANDARD_BUTTONS = {
"save": ActionButton(
id="save",
icon="💾",
tooltip="Save (Alt+S)",
),
"close": ActionButton(
id="close",
icon="",
tooltip="Close",
),
"minimize": ActionButton(
id="minimize",
icon="",
tooltip="Minimize",
),
"maximize": ActionButton(
id="maximize",
icon="",
tooltip="Maximize",
),
"restore": ActionButton(
id="restore",
icon="",
tooltip="Restore",
visible=False, # Hidden by default, shown when maximized
),
"split": ActionButton(
id="split",
icon="",
tooltip="Split terminal",
),
}
+88 -2
View File
@@ -9,6 +9,37 @@ from textual.widgets import Static, TextArea
from clide.models.editor import CursorPosition, FileBuffer
# TypeScript highlight query (basic)
TYPESCRIPT_HIGHLIGHTS = """
(comment) @comment
(string) @string
(number) @number
(identifier) @variable
(type_identifier) @type
(property_identifier) @property
(function_declaration name: (identifier) @function)
(method_definition name: (property_identifier) @function.method)
(call_expression function: (identifier) @function.call)
(import_statement) @keyword
(export_statement) @keyword
["const" "let" "var" "function" "class" "interface" "type" "enum"
"if" "else" "for" "while" "do" "switch" "case" "default" "break"
"continue" "return" "throw" "try" "catch" "finally" "new" "delete"
"typeof" "instanceof" "in" "of" "async" "await" "yield" "import"
"export" "from" "as" "extends" "implements" "static" "public"
"private" "protected" "readonly" "abstract" "declare" "namespace"
"module" "require"] @keyword
["=>" "=" "+" "-" "*" "/" "%" "**" "++" "--" "==" "!=" "===" "!=="
"<" ">" "<=" ">=" "&&" "||" "!" "?" ":" "?." "??" "&" "|" "^" "~"
"<<" ">>" ">>>"] @operator
["(" ")" "[" "]" "{" "}"] @punctuation.bracket
["," "." ";" ":"] @punctuation.delimiter
(true) @constant.builtin
(false) @constant.builtin
(null) @constant.builtin
(undefined) @constant.builtin
"""
class EditorPane(Vertical):
"""Editor pane with TextArea and status bar."""
@@ -72,6 +103,45 @@ class EditorPane(Vertical):
def __init__(self, buffer: FileBuffer | None = None, **kwargs) -> None:
super().__init__(**kwargs)
self._buffer = buffer
self._typescript_registered = False
def on_mount(self) -> None:
"""Register additional languages on mount."""
self._register_typescript()
def _register_typescript(self) -> None:
"""Register TypeScript and TSX languages if available."""
if self._typescript_registered:
return
try:
import tree_sitter_typescript as tst
textarea = self.query_one("#editor-textarea", TextArea)
# Register TypeScript
try:
textarea.register_language(
"typescript",
tst.language_typescript(),
TYPESCRIPT_HIGHLIGHTS,
)
except Exception:
pass
# Register TSX (reuse TypeScript highlights)
try:
textarea.register_language(
"tsx",
tst.language_tsx(),
TYPESCRIPT_HIGHLIGHTS,
)
except Exception:
pass
self._typescript_registered = True
except ImportError:
pass
def compose(self) -> ComposeResult:
if self._buffer:
@@ -97,8 +167,19 @@ class EditorPane(Vertical):
self._buffer = buffer
textarea = self.query_one("#editor-textarea", TextArea)
# Register TypeScript if needed
if buffer.language in ("typescript", "tsx"):
self._register_typescript()
textarea.load_text(buffer.content)
textarea.language = buffer.language
# Set language (will use JavaScript as fallback for TS if registration failed)
language = buffer.language
if language in ("typescript", "tsx") and language not in textarea.available_languages:
language = "javascript" # Fallback to JS highlighting
textarea.language = language
# Update tab
tab = self.query_one(".file-tab", Static)
@@ -167,7 +248,12 @@ class EditorPane(Vertical):
"""Load a file from disk into the editor."""
from clide.services.file_service import FileService
content = FileService.read_file(path)
try:
content = FileService.read_file(path)
except OSError as e:
# Show error in editor
content = f"# Error loading file\n# {e}"
language = FileService.detect_language(path)
buffer = FileBuffer(
+42
View File
@@ -85,3 +85,45 @@ class FilesView(DirectoryTree):
def refresh_tree(self) -> None:
"""Refresh the directory tree."""
self.reload()
def highlight_path(self, path: Path) -> None:
"""Highlight a path in the tree (expand parents and scroll to it).
Used to show which file Claude is working with.
"""
# Normalize the path
try:
path = path.resolve()
except Exception:
return
# Find and select the node
def find_node(node, target_path):
"""Recursively find a node by path."""
if node.data and hasattr(node.data, 'path'):
if node.data.path.resolve() == target_path:
return node
for child in node.children:
result = find_node(child, target_path)
if result:
return result
return None
target_node = find_node(self.root, path)
if target_node:
# Expand all parent nodes
parent = target_node.parent
while parent:
parent.expand()
parent = parent.parent
# Select and scroll to the node
self.select_node(target_node)
self.scroll_to_node(target_node)
def on_directory_tree_file_selected(
self, event: DirectoryTree.FileSelected
) -> None:
"""Re-emit file selection as FilesView.FileSelected."""
event.stop()
self.post_message(self.FileSelected(node=event.node, path=event.path))
+368 -17
View File
@@ -3,9 +3,11 @@
from __future__ import annotations
import asyncio
import codecs
import fcntl
import os
import pty
import re
import shutil
import struct
import termios
@@ -13,7 +15,6 @@ import time
from pathlib import Path
from typing import TYPE_CHECKING
import pyte
from rich.text import Text
from textual.containers import Vertical
from textual.message import Message
@@ -21,10 +22,38 @@ from textual.reactive import reactive
from textual.strip import Strip
from textual.widget import Widget
# Use vendored pyte with diagnostic logging support
from clide.vendor import pyte
from clide.services.settings_service import get_settings_service
if TYPE_CHECKING:
from textual.app import ComposeResult
def _setup_terminal_debug_logging() -> None:
"""Set up terminal debug logging if enabled in settings."""
settings = get_settings_service()
if not settings.get("terminal_debug", False):
return
# Create log file in settings directory
log_path = settings.settings_dir / "terminal_debug.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
# Open log file (append mode)
log_file = open(log_path, "a", encoding="utf-8")
def debug_logger(message: str) -> None:
"""Log debug message with timestamp."""
import datetime
timestamp = datetime.datetime.now().isoformat()
log_file.write(f"[{timestamp}] {message}\n")
log_file.flush()
# Set up pyte debug logging
pyte.set_debug_logger(debug_logger)
class TerminalDisplay(Widget, can_focus=True):
"""A terminal emulator widget using pyte."""
@@ -36,6 +65,10 @@ class TerminalDisplay(Widget, can_focus=True):
}
"""
# Internal padding (rendered as part of terminal content, uses terminal bg)
PADDING_LEFT = 1
PADDING_RIGHT = 1
def __init__(
self,
cols: int = 80,
@@ -45,7 +78,12 @@ class TerminalDisplay(Widget, can_focus=True):
super().__init__(**kwargs)
self._cols = cols
self._rows = rows
self._screen = pyte.Screen(cols, rows)
# Set up debug logging before creating pyte objects
_setup_terminal_debug_logging()
self._screen = pyte.HistoryScreen(cols, rows, history=1000)
self._screen.set_mode(pyte.modes.LNM) # Line feed mode
self._stream = pyte.Stream(self._screen)
self._master_fd: int | None = None
self._pid: int | None = None
@@ -53,32 +91,96 @@ class TerminalDisplay(Widget, can_focus=True):
self._refresh_task: asyncio.Task | None = None
self._needs_refresh: bool = False
self._last_refresh: float = 0
self._pending_start: tuple[str, str] | None = None
# Incremental UTF-8 decoder to handle partial sequences at buffer boundaries
self._decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
# Scroll offset for viewing history (0 = at bottom/current, positive = scrolled up)
self._scroll_offset: int = 0
def on_mount(self) -> None:
"""Initialize terminal size from widget dimensions."""
# Get actual widget size, accounting for internal padding
size = self.size
if size.width > 0 and size.height > 0:
self._cols = max(size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20)
self._rows = size.height
self._screen.resize(self._rows, self._cols)
# If start was called before mount, do it now
if self._pending_start:
command, cwd = self._pending_start
self._pending_start = None
self._do_start(command, cwd)
def on_resize(self, event) -> None:
"""Handle terminal resize."""
# Get new size from widget
new_cols = max(event.size.width, 20)
new_rows = max(event.size.height, 5)
# Get new size from widget, accounting for internal padding
new_cols = max(self.size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20)
new_rows = max(self.size.height, 5)
if new_cols != self._cols or new_rows != self._rows:
old_cols = self._cols
# Update dimensions first
self._cols = new_cols
self._rows = new_rows
# Resize pyte screen - this will preserve content where possible
self._screen.resize(new_rows, new_cols)
# Update PTY size if running
# If screen got wider, clear the new columns to avoid stale data
# pyte's resize should handle this, but let's be defensive
if new_cols > old_cols:
for y in range(new_rows):
line = self._screen.buffer[y]
for x in range(old_cols, new_cols):
# Clear any stale data in new columns
line[x] = pyte.screens.Char(" ")
# Update PTY size if running - Claude will redraw
if self._master_fd is not None:
self._set_pty_size(self._master_fd, new_rows, new_cols)
# Force a full refresh
self.refresh()
def _set_pty_size(self, fd: int, rows: int, cols: int) -> None:
"""Set the PTY window size."""
"""Set the PTY window size and notify the child process."""
try:
winsize = struct.pack("HHHH", rows, cols, 0, 0)
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
# Send SIGWINCH to notify the child process of resize
if self._pid is not None:
import signal
try:
os.kill(self._pid, signal.SIGWINCH)
except OSError:
pass
except OSError:
pass
def start(self, command: str, cwd: str) -> None:
"""Start a process in the terminal."""
# If not mounted yet, defer start
if not self.is_mounted:
self._pending_start = (command, cwd)
return
self._do_start(command, cwd)
def _do_start(self, command: str, cwd: str) -> None:
"""Actually start the process in the terminal."""
# Reset decoder state for new process
self._decoder.reset()
# Get current widget size, accounting for internal padding
size = self.size
if size.width > 0 and size.height > 0:
self._cols = max(size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20)
self._rows = size.height
self._screen.resize(self._rows, self._cols)
# Fork a PTY
pid, master_fd = pty.fork()
@@ -87,6 +189,8 @@ class TerminalDisplay(Widget, can_focus=True):
os.chdir(cwd)
os.environ["TERM"] = "xterm-256color"
os.environ["COLORTERM"] = "truecolor"
os.environ["COLUMNS"] = str(self._cols)
os.environ["LINES"] = str(self._rows)
os.execlp(command, command)
else:
# Parent process
@@ -123,6 +227,21 @@ class TerminalDisplay(Widget, can_focus=True):
# Small delay to batch rapid updates
await asyncio.sleep(0.008)
# Regex to filter escape sequences that pyte doesn't handle
# Kitty keyboard protocol, bracketed paste mode queries, etc.
_UNSUPPORTED_ESCAPES = re.compile(
r"\x1b\[[\=\>\<][0-9;]*[a-zA-Z]" # Kitty keyboard protocol (=, >, or < prefix)
r"|\x1b\[\?[0-9;]*u" # Kitty keyboard query
r"|\x1b\[\?[0-9;]*c" # Device attributes query
r"|\x1b\[>[0-9;]*c" # Secondary device attributes
r"|\x1b\]\d+;[^\x07\x1b]*(?:\x07|\x1b\\)" # OSC sequences (title, etc.)
r"|\x1b\[\?2026[hl]" # Synchronized update mode (not used by pyte)
)
def _filter_unsupported_escapes(self, data: str) -> str:
"""Filter out escape sequences that pyte doesn't handle."""
return self._UNSUPPORTED_ESCAPES.sub("", data)
async def _read_output(self) -> None:
"""Read output from the PTY."""
if self._master_fd is None:
@@ -138,8 +257,15 @@ class TerminalDisplay(Widget, can_focus=True):
if not data:
break
# Use incremental decoder to handle partial UTF-8 sequences
# at buffer boundaries (prevents box-drawing chars getting corrupted)
text = self._decoder.decode(data)
if not text:
continue # Still waiting for more bytes to complete a sequence
text = self._filter_unsupported_escapes(text)
# Feed data to pyte
self._stream.feed(data.decode("utf-8", errors="replace"))
self._stream.feed(text)
self._schedule_refresh()
except BlockingIOError:
@@ -201,42 +327,200 @@ class TerminalDisplay(Widget, can_focus=True):
except OSError:
pass
# ANSI 256-color palette (standard 16 colors)
ANSI_COLORS = {
"black": "#000000",
"red": "#cd0000",
"green": "#00cd00",
"yellow": "#cdcd00",
"blue": "#0000ee",
"magenta": "#cd00cd",
"cyan": "#00cdcd",
"white": "#e5e5e5",
"brightblack": "#7f7f7f",
"brightred": "#ff0000",
"brightgreen": "#00ff00",
"brightyellow": "#ffff00",
"brightblue": "#5c5cff",
"brightmagenta": "#ff00ff",
"brightcyan": "#00ffff",
"brightwhite": "#ffffff",
}
def _convert_color(self, color: str, is_bg: bool = False) -> str | None:
"""Convert pyte color to Rich color string."""
if not color or color == "default":
return None
# Handle hex colors (with or without # prefix)
if color.startswith("#"):
return color
# Check if it's a hex color without # (pyte returns "ff0000" not "#ff0000")
if len(color) == 6 and all(c in "0123456789abcdefABCDEF" for c in color):
return f"#{color}"
# Handle named ANSI colors
color_lower = color.lower()
if color_lower in self.ANSI_COLORS:
return self.ANSI_COLORS[color_lower]
# Handle 256-color palette (numeric)
try:
num = int(color)
if 0 <= num <= 255:
return f"color({num})"
except ValueError:
pass
# Fallback: try to use the color name directly
return color
def _get_line_at(self, y: int):
"""Get the line at display position y, accounting for scroll offset.
When scrolled, we show lines from history mixed with current buffer.
scroll_offset=0 means showing current screen.
scroll_offset=N means the top of display shows N lines back in history.
"""
history = self._screen.history.top
history_len = len(history)
if self._scroll_offset == 0:
# Not scrolled - show current buffer
return self._screen.buffer[y]
# Calculate which line to show
# Display line 0 should show history[history_len - scroll_offset]
# Display line N should show history[history_len - scroll_offset + N]
# If that index >= history_len, we're into the current buffer
history_index = history_len - self._scroll_offset + y
if history_index < 0:
# Before start of history - return blank
return None
elif history_index < history_len:
# In history
return history[history_index]
else:
# In current buffer
buffer_index = history_index - history_len
if buffer_index < self._rows:
return self._screen.buffer[buffer_index]
return None
def render_line(self, y: int) -> Strip:
"""Render a line of the terminal."""
if y >= self._rows:
return Strip.blank(self._cols)
# Screen buffer dimensions (what pyte has)
screen_cols = self._cols
screen_rows = self._rows
# Widget display dimensions (what we need to output)
output_width = self.size.width if self.size.width > 0 else screen_cols + self.PADDING_LEFT + self.PADDING_RIGHT
if y >= screen_rows:
return Strip.blank(output_width)
# Get the line to render, accounting for scroll offset
line = self._get_line_at(y)
if line is None:
return Strip.blank(output_width)
line = self._screen.buffer[y]
text = Text()
for x in range(self._cols):
# Add left padding (uses terminal background)
text.append(" " * self.PADDING_LEFT)
# Only access indices within the screen buffer
cols_to_render = min(screen_cols, output_width - self.PADDING_LEFT - self.PADDING_RIGHT)
# Debug logging for render pipeline
debug_logger = getattr(pyte, '_debug_logger', None)
log_this_line = False
for x in range(cols_to_render):
char = line[x]
char_data = char.data if char.data else " "
# Check for box-drawing and other potentially problematic characters
if len(char_data) == 1:
code = ord(char_data)
# Log box-drawing characters (U+2500-U+257F)
if 0x2500 <= code <= 0x257F:
log_this_line = True
if debug_logger:
debug_logger(f"RENDER y={y} x={x}: box-drawing U+{code:04X} char='{char_data}'")
# Handle characters that may not render correctly
if len(char_data) == 1:
code = ord(char_data)
# Control characters (except space)
if code < 32 and code != 0:
char_data = " "
# DEL and C1 control characters
elif 127 <= code <= 159:
char_data = " "
# Braille patterns (U+2800-U+28FF) - used for spinners
# Replace with simple ASCII spinner chars or spaces
elif 0x2800 <= code <= 0x28FF:
# Map braille spinner to simple dots
char_data = "·"
# Build style from pyte character attributes
style_parts = []
if char.fg and char.fg != "default":
style_parts.append(f"color({char.fg})" if char.fg.startswith("#") else char.fg)
# Foreground color
fg = self._convert_color(char.fg)
if fg:
style_parts.append(fg)
if char.bg and char.bg != "default":
style_parts.append(f"on color({char.bg})" if char.bg.startswith("#") else f"on {char.bg}")
# Background color
bg = self._convert_color(char.bg, is_bg=True)
if bg:
style_parts.append(f"on {bg}")
# Text attributes
if char.bold:
style_parts.append("bold")
if char.italics:
style_parts.append("italic")
if char.underscore:
style_parts.append("underline")
if char.strikethrough:
style_parts.append("strike")
if char.reverse:
style_parts.append("reverse")
style = " ".join(style_parts) if style_parts else None
text.append(char_data, style=style)
# Pad with spaces to fill remaining width (includes right padding)
content_width = self.PADDING_LEFT + cols_to_render
if output_width > content_width:
text.append(" " * (output_width - content_width))
# Debug: log the Rich Text content before rendering
if log_this_line and debug_logger:
debug_logger(f"RENDER y={y}: Rich Text plain='{text.plain[:80]}...'")
# Render text to segments for Strip
segments = list(text.render(self.app.console))
return Strip(segments)
# Debug: log segments if we had box-drawing chars
if log_this_line and debug_logger:
for i, seg in enumerate(segments[:10]): # First 10 segments
seg_text = seg.text if hasattr(seg, 'text') else str(seg)
if len(seg_text) <= 5:
debug_logger(f"RENDER y={y} seg[{i}]: '{seg_text}' (repr: {repr(seg_text)})")
strip = Strip(segments)
# Ensure strip is exactly the output width
if strip.cell_length != output_width:
strip = strip.crop_extend(0, output_width, None)
return strip
def on_key(self, event) -> None:
"""Handle key presses."""
@@ -277,11 +561,78 @@ class TerminalDisplay(Widget, can_focus=True):
self.send("\x0c")
event.prevent_default()
event.stop()
elif event.key == "shift+pageup":
# Scroll up in history
self._scroll_up(self._rows // 2)
event.prevent_default()
event.stop()
elif event.key == "shift+pagedown":
# Scroll down in history
self._scroll_down(self._rows // 2)
event.prevent_default()
event.stop()
elif event.key == "shift+home":
# Scroll to top of history
self._scroll_to_top()
event.prevent_default()
event.stop()
elif event.key == "shift+end":
# Scroll to bottom (current)
self._scroll_to_bottom()
event.prevent_default()
event.stop()
elif event.character and len(event.character) == 1:
# Any typing scrolls to bottom
self._scroll_to_bottom()
self.send(event.character)
event.prevent_default()
event.stop()
def on_mouse_scroll_up(self, event) -> None:
"""Handle mouse scroll up (view older content)."""
self._scroll_up(3)
event.prevent_default()
event.stop()
def on_mouse_scroll_down(self, event) -> None:
"""Handle mouse scroll down (view newer content)."""
self._scroll_down(3)
event.prevent_default()
event.stop()
def _scroll_up(self, lines: int) -> None:
"""Scroll up (back in history) by given number of lines."""
max_scroll = len(self._screen.history.top)
new_offset = min(self._scroll_offset + lines, max_scroll)
if new_offset != self._scroll_offset:
self._scroll_offset = new_offset
self.refresh()
def _scroll_down(self, lines: int) -> None:
"""Scroll down (forward toward current) by given number of lines."""
new_offset = max(self._scroll_offset - lines, 0)
if new_offset != self._scroll_offset:
self._scroll_offset = new_offset
self.refresh()
def _scroll_to_top(self) -> None:
"""Scroll to the top of history."""
max_scroll = len(self._screen.history.top)
if self._scroll_offset != max_scroll:
self._scroll_offset = max_scroll
self.refresh()
def _scroll_to_bottom(self) -> None:
"""Scroll to the bottom (current screen)."""
if self._scroll_offset != 0:
self._scroll_offset = 0
self.refresh()
@property
def history_size(self) -> int:
"""Get the number of lines in scrollback history."""
return len(self._screen.history.top)
class ClaudePanel(Vertical):
"""Terminal panel running Claude Code CLI.
+11
View File
@@ -132,6 +132,17 @@ class SidebarPanel(Vertical):
except Exception:
pass
def highlight_file(self, path: Path) -> None:
"""Highlight a file in the file browser.
Used to show which file Claude is working with.
"""
try:
files_view = self.query_one(FilesView)
files_view.highlight_path(path)
except Exception:
pass
def focus_tab(self, tab_id: str) -> None:
"""Focus a specific tab."""
tabs = self.query_one("#sidebar-tabs", TabbedContent)
+214 -12
View File
@@ -3,12 +3,13 @@
from pathlib import Path
from textual.app import ComposeResult
from textual.containers import Vertical
from textual.containers import Horizontal, Vertical
from textual.message import Message
from textual.reactive import reactive
from textual.widgets import TabbedContent, TabPane
from textual.widgets import ContentSwitcher, Tab, TabPane, Tabs
from clide.models.diff import DiffContent
from clide.widgets.components.action_bar import ActionBar, ActionButton, STANDARD_BUTTONS
from clide.widgets.components.diff_pane import DiffPane
from clide.widgets.components.editor_pane import EditorPane
from clide.widgets.components.terminal_pane import TerminalPane
@@ -33,11 +34,41 @@ class WorkspacePanel(Vertical):
display: none;
}
WorkspacePanel TabbedContent {
WorkspacePanel.maximized {
height: 100%;
}
WorkspacePanel TabPane {
/* Header row with tabs and action bar */
WorkspacePanel #workspace-header {
height: auto;
width: 100%;
background: $surface;
}
WorkspacePanel #workspace-header Tabs {
width: 1fr;
}
WorkspacePanel #workspace-header #workspace-action-bar {
width: auto;
height: auto;
padding: 0 1;
}
WorkspacePanel #workspace-header ActionBarButton {
height: 1;
min-width: 3;
margin: 0;
padding: 0;
}
WorkspacePanel ContentSwitcher {
height: 1fr;
}
WorkspacePanel #pane-editor,
WorkspacePanel #pane-diff,
WorkspacePanel #pane-terminal {
height: 100%;
padding: 0;
}
@@ -78,6 +109,16 @@ class WorkspacePanel(Vertical):
# Reactive state - persisted when hidden
visible: reactive[bool] = reactive(False)
active_tab: reactive[str] = reactive("editor")
maximized: reactive[bool] = reactive(False)
# Messages for app-level actions
class MaximizeRequested(Message):
"""Emitted when workspace should be maximized."""
pass
class RestoreRequested(Message):
"""Emitted when workspace should be restored from maximized."""
pass
def __init__(
self,
@@ -87,19 +128,170 @@ class WorkspacePanel(Vertical):
super().__init__(**kwargs)
self._workdir = workdir or Path.cwd()
self.id = "panel-workspace"
self._action_bar: ActionBar | None = None
def compose(self) -> ComposeResult:
with TabbedContent(id="workspace-tabs"):
with TabPane("Editor", id="workspace-editor"):
# Header row with tabs and action bar
with Horizontal(id="workspace-header"):
yield Tabs(
Tab("Editor", id="tab-editor"),
Tab("Diff", id="tab-diff"),
Tab("Terminal", id="tab-terminal"),
id="workspace-tabs",
)
self._action_bar = ActionBar(id="workspace-action-bar")
yield self._action_bar
# Content area
with ContentSwitcher(id="workspace-content", initial="pane-editor"):
with Vertical(id="pane-editor"):
yield EditorPane(id="editor-pane")
with TabPane("Diff", id="workspace-diff"):
with Vertical(id="pane-diff"):
yield DiffPane(id="diff-pane")
with TabPane("Terminal", id="workspace-terminal"):
with Vertical(id="pane-terminal"):
yield TerminalPane(cwd=self._workdir, id="terminal-pane")
def on_mount(self) -> None:
"""Set initial visibility."""
"""Set initial visibility and configure action bar."""
self._update_visibility()
self._setup_action_bar()
def _setup_action_bar(self) -> None:
"""Set up the action bar with standard buttons."""
if self._action_bar is None:
return
# Register standard buttons using simple ASCII icons
# Save button (for editor)
self._action_bar.register_button(ActionButton(
id="save",
icon="[S]",
tooltip="Save",
))
# Add separator
self._action_bar.add_separator()
# Close button
self._action_bar.register_button(ActionButton(
id="close",
icon="x",
tooltip="Close",
))
# Minimize button
self._action_bar.register_button(ActionButton(
id="minimize",
icon="_",
tooltip="Minimize",
))
# Maximize button
self._action_bar.register_button(ActionButton(
id="maximize",
icon="^",
tooltip="Maximize",
))
# Restore button (hidden by default)
self._action_bar.register_button(ActionButton(
id="restore",
icon="v",
tooltip="Restore",
visible=False,
))
# Update button visibility based on current tab
self._update_action_bar_for_tab(self.active_tab)
def _update_action_bar_for_tab(self, tab: str) -> None:
"""Update action bar buttons based on active tab."""
if self._action_bar is None:
return
# Save button only visible for editor
self._action_bar.set_button_visible("save", tab == "editor")
# Update save button enabled state based on editor modified state
if tab == "editor":
try:
editor = self.query_one("#editor-pane", EditorPane)
self._action_bar.set_button_enabled("save", editor.modified)
except Exception:
self._action_bar.set_button_enabled("save", False)
def watch_maximized(self, maximized: bool) -> None:
"""Handle maximize state changes."""
if self._action_bar is None:
return
# Toggle maximize/restore button visibility
self._action_bar.set_button_visible("maximize", not maximized)
self._action_bar.set_button_visible("restore", maximized)
# Update CSS class
if maximized:
self.add_class("maximized")
else:
self.remove_class("maximized")
def on_action_bar_button_pressed(self, event: ActionBar.ButtonPressed) -> None:
"""Handle action bar button presses."""
button_id = event.button_id
if button_id == "save":
self._action_save()
elif button_id == "close":
self._action_close()
elif button_id == "minimize":
self._action_minimize()
elif button_id == "maximize":
self._action_maximize()
elif button_id == "restore":
self._action_restore()
def _action_save(self) -> None:
"""Save the current file in editor."""
try:
editor = self.query_one("#editor-pane", EditorPane)
editor.save()
except Exception as e:
self.app.notify(f"Save failed: {e}", severity="error")
def _action_close(self) -> None:
"""Close the workspace panel."""
self.post_message(self.CloseRequested())
self.hide()
def _action_minimize(self) -> None:
"""Minimize (hide) the workspace panel."""
self.hide()
def _action_maximize(self) -> None:
"""Maximize the workspace panel."""
self.maximized = True
self.post_message(self.MaximizeRequested())
def _action_restore(self) -> None:
"""Restore from maximized state."""
self.maximized = False
self.post_message(self.RestoreRequested())
def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None:
"""Handle tab switches to update action bar and content."""
# Extract tab name from tab id (e.g., "tab-editor" -> "editor")
tab_id = event.tab.id
if tab_id and tab_id.startswith("tab-"):
tab_name = tab_id.replace("tab-", "")
self.active_tab = tab_name
self._update_action_bar_for_tab(tab_name)
# Switch content
try:
content = self.query_one("#workspace-content", ContentSwitcher)
content.current = f"pane-{tab_name}"
except Exception:
pass
def watch_visible(self, visible: bool) -> None:
"""Handle visibility changes - hide, don't destroy."""
@@ -128,9 +320,19 @@ class WorkspacePanel(Vertical):
def focus_tab(self, tab_id: str) -> None:
"""Focus a specific tab."""
tabs = self.query_one("#workspace-tabs", TabbedContent)
tabs.active = f"workspace-{tab_id}"
self.active_tab = tab_id
try:
# Activate the tab
tabs = self.query_one("#workspace-tabs", Tabs)
tabs.active = f"tab-{tab_id}"
# Switch content
content = self.query_one("#workspace-content", ContentSwitcher)
content.current = f"pane-{tab_id}"
self.active_tab = tab_id
self._update_action_bar_for_tab(tab_id)
except Exception:
pass
# Editor methods
def open_file(self, path: Path, line: int | None = None) -> None:
+18 -1
View File
@@ -10,7 +10,7 @@ readme = "README.md"
requires-python = ">=3.12"
license = "MIT"
authors = [
{ name = "Your Name", email = "you@example.com" }
{ name = "Jeroen Schweitzer", email = "you@example.com" }
]
classifiers = [
"Development Status :: 3 - Alpha",
@@ -30,6 +30,23 @@ dependencies = [
"pydantic-settings>=2.0.0",
"pluggy>=1.4.0",
"rich>=13.0.0",
"watchdog>=4.0.0",
"wcwidth>=0.2.0", # Required by vendored pyte
# Tree-sitter for syntax highlighting
"tree-sitter>=0.21.0",
"tree-sitter-python>=0.21.0",
"tree-sitter-javascript>=0.21.0",
"tree-sitter-typescript>=0.21.0",
"tree-sitter-html>=0.21.0",
"tree-sitter-css>=0.21.0",
"tree-sitter-json>=0.21.0",
"tree-sitter-yaml>=0.5.0",
"tree-sitter-toml>=0.5.0",
"tree-sitter-markdown>=0.2.0",
"tree-sitter-rust>=0.21.0",
"tree-sitter-go>=0.21.0",
"tree-sitter-java>=0.21.0",
"tree-sitter-bash>=0.21.0",
]
[project.optional-dependencies]