"""
Basic Output Buffer
Manages terminal output with scrollback and line tracking
"""
import logging
from collections import deque
from typing import List, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class OutputLine:
"""Represents a single line of terminal output"""
def __init__(self, text: str, timestamp: Optional[datetime] = None):
self.text = text
self.timestamp = timestamp or datetime.now()
self.marked = False # For marking lines to send to Claude
def __str__(self) -> str:
return self.text
def __repr__(self) -> str:
return f"OutputLine('{self.text[:50]}...', marked={self.marked})"
class OutputBuffer:
"""
Manages terminal output with scrollback buffer
"""
def __init__(self, max_lines: int = 1000):
"""
Initialize Output Buffer
Args:
max_lines: Maximum number of lines to keep in buffer
"""
self.max_lines = max_lines
self.lines: deque[OutputLine] = deque(maxlen=max_lines)
self.current_output = "" # Accumulates output until newline
def add(self, text: str) -> List[OutputLine]:
"""
Add text to buffer
Args:
text: Text to add (may contain multiple lines)
Returns:
List of newly created OutputLine objects
"""
new_lines = []
self.current_output += text
# Process complete lines
while '\n' in self.current_output:
line_text, self.current_output = self.current_output.split('\n', 1)
line = OutputLine(line_text)
self.lines.append(line)
new_lines.append(line)
return new_lines
def flush(self) -> Optional[OutputLine]:
"""
Flush any remaining partial line
Returns:
OutputLine if there was partial content, None otherwise
"""
if self.current_output:
line = OutputLine(self.current_output)
self.lines.append(line)
self.current_output = ""
return line
return None
def get_last_n(self, n: int = 100) -> List[OutputLine]:
"""
Get last N lines from buffer
Args:
n: Number of lines to retrieve
Returns:
List of OutputLine objects
"""
return list(self.lines)[-n:]
def get_all(self) -> List[OutputLine]:
"""Get all lines in buffer"""
return list(self.lines)
def get_text(self, start: int = 0, end: Optional[int] = None) -> str:
"""
Get text from buffer
Args:
start: Start line index
end: End line index (None for all)
Returns:
Concatenated text
"""
lines_list = list(self.lines)[start:end]
return '\n'.join(line.text for line in lines_list)
def clear(self) -> None:
"""Clear all buffer contents"""
self.lines.clear()
self.current_output = ""
logger.debug("Output buffer cleared")
def mark_lines(self, start: int, end: int) -> int:
"""
Mark lines for Claude analysis
Args:
start: Start line index
end: End line index
Returns:
Number of lines marked
"""
lines_list = list(self.lines)
count = 0
for i in range(start, min(end, len(lines_list))):
if i >= 0:
lines_list[i].marked = True
count += 1
logger.debug(f"Marked {count} lines")
return count
def unmark_all(self) -> None:
"""Unmark all lines"""
for line in self.lines:
line.marked = False
def get_marked(self) -> List[OutputLine]:
"""Get all marked lines"""
return [line for line in self.lines if line.marked]
def get_stats(self) -> dict:
"""
Get buffer statistics
Returns:
Dictionary with statistics
"""
return {
'total_lines': len(self.lines),
'max_lines': self.max_lines,
'marked_lines': sum(1 for line in self.lines if line.marked),
'partial_line_length': len(self.current_output)
}
