logloglog
LogLogLog - Efficient scrollback indexing for large log files.
configure_logging
def configure_logging(level=logging.INFO)
Configure logging for LogLogLog.
logloglog.line_index
Simple line indexing with periodic summaries for efficient wrapping calculations.
MAX_WIDTH
Maximum terminal width we support
SUMMARY_INTERVAL
Store summary every N lines
LineIndex Objects
class LineIndex()
Indexes log lines with byte positions, widths, and periodic summaries.
Stores:
- line_positions: byte offset of each line in the log file
- line_widths: display width of each line
- summaries: every 1000 lines, cumulative display rows for each width 1-512
__init__
def __init__(index_path: Path)
Initialize line index with given path.
open
def open(create: bool = False)
Open index files.
close
def close()
Close all index files.
append_line
def append_line(position: int, width: int)
Append a new line to the index.
Arguments:
position- Byte offset of line start in log filewidth- Display width of the line
get_line_position
def get_line_position(line_no: int) -> int
Get byte position of a line.
get_line_width
def get_line_width(line_no: int) -> int
Get display width of a line.
get_total_display_rows
def get_total_display_rows(width: int) -> int
Get total display rows for all lines at given terminal width.
Arguments:
width- Terminal width
Returns:
Total number of display rows
get_display_row_for_line
def get_display_row_for_line(line_no: int, width: int) -> int
Get the display row number where a logical line starts.
Arguments:
line_no- Logical line numberwidth- Terminal width
Returns:
Display row number where this line starts
get_line_for_display_row
def get_line_for_display_row(display_row: int, width: int) -> Tuple[int, int]
Find the logical line containing the given display row.
Arguments:
display_row- Display row to findwidth- Terminal width
Returns:
Tuple of (line_number, row_offset_within_line)
__len__
def __len__() -> int
Get total number of indexed lines.
logloglog.widthview
WidthView class for viewing logs at a specific terminal width.
WidthView Objects
class WidthView()
A width-specific view of a LogLogLog using Python container protocols.
__init__
def __init__(logloglog: "LogLogLog", width: int)
Initialize a WidthView.
Arguments:
logloglog- The LogLogLog instance to viewwidth- Terminal width for wrapping
line_at
def line_at(row: int) -> Tuple[int, int]
Find which logical line contains the given display row.
Arguments:
row- Display row number
Returns:
Tuple of (line_number, offset_within_line)
Raises:
IndexError- If row is out of bounds
row_for
def row_for(line_no: int) -> int
Get the display row where a logical line starts.
Arguments:
line_no- Logical line number
Returns:
Display row number
__getitem__
def __getitem__(row_no: int) -> str
Get text at display row.
Arguments:
row_no- Display row number (negative indexing supported)
Returns:
Text at the display row (may be partial line if wrapped)
Raises:
IndexError- If row_no is out of bounds
__len__
def __len__() -> int
Get total number of display rows.
__iter__
def __iter__() -> Iterator[str]
Iterate over display rows.
logloglog.ui
logloglog.ui.textual.log_widget
LogWidget Objects
class LogWidget(ScrollView)
A scrollable widget to display log data.
LogUpdated Objects
class LogUpdated(Message)
Posted when log display updates (scroll, resize, etc).
on_mount
def on_mount()
Called when widget is mounted.
on_resize
def on_resize(event)
Called when widget is resized.
set_width
def set_width(width: int)
Update the view width and preserve scroll position.
watch_scroll_y
def watch_scroll_y(old_value: float, new_value: float) -> None
Called when scroll position changes.
watch_virtual_size
def watch_virtual_size(old_size: Size, new_size: Size) -> None
Called when virtual (scrollable) size changes.
render_line
def render_line(y: int) -> Strip
Render a single line of the log.
scroll_to
def scroll_to(x=None, y=None, **kwargs)
Override scroll_to to always disable animation.
scroll_up
def scroll_up(**kwargs)
Override scroll_up to always disable animation.
scroll_down
def scroll_down(**kwargs)
Override scroll_down to always disable animation.
start_auto_refresh
def start_auto_refresh(interval: float = 1.0)
Start background task to automatically refresh log data.
stop_auto_refresh
def stop_auto_refresh()
Stop the background refresh task.
enable_auto_refresh
def enable_auto_refresh(enabled: bool = True)
Enable or disable auto-refresh functionality.
arefresh_log_data
async def arefresh_log_data()
Async method to refresh log data without blocking the UI.
aset_width
async def aset_width(width: int)
Async version of set_width for non-blocking width changes.
on_unmount
def on_unmount()
Called when widget is unmounted - cleanup background tasks.
action_scroll_home
def action_scroll_home()
Jump to the start of the log.
action_scroll_end
def action_scroll_end()
Jump to the end of the log.
logloglog.ui.textual
logloglog.tools.stream_logs
Stream system logs for logloglog.
This tool discovers and streams system logs from /var/log, handling both historical logs (sorted by creation time) and live log following.
discover_historical_logs
def discover_historical_logs() -> Iterator[Tuple[float, Path]]
Discover all log files in /var/log sorted by creation time.
Yields:
Tuple of (creation_time, filepath) sorted by creation time (oldest first)
discover_live_logs
def discover_live_logs(last_modified_minutes: int = 60) -> List[Path]
Discover .log files that were modified within the last N minutes.
Arguments:
last_modified_minutes- Only include files modified within this many minutes
Returns:
List of .log file paths
is_text_file
def is_text_file(filepath: Path) -> bool
Check if a file is a text file using the file command.
Arguments:
filepath- Path to check
Returns:
True if the file appears to be text
stream_file_content
def stream_file_content(filepath: Path) -> None
Stream the content of a file to stdout, handling compression.
Arguments:
filepath- File to stream
stream_historical_logs
def stream_historical_logs() -> None
Stream all historical log files to stdout.
follow_live_logs
async def follow_live_logs(last_modified_minutes: int = 60) -> None
Follow live log files using pure Python implementation.
Arguments:
last_modified_minutes- Only follow files modified within this many minutes
tail_multiple_files
async def tail_multiple_files(filepaths: List[Path]) -> None
Pure Python implementation of tail -F for multiple files.
Arguments:
filepaths- List of file paths to tail
setup_signal_handlers
def setup_signal_handlers() -> None
Set up clean signal handling.
main
async def main() -> None
Main entry point.
logloglog.tools
Tools subpackage for logloglog utilities.
logloglog.__main__
Entry point for the package
logloglog.cache
Cache management for LogLogLog.
Cache Objects
class Cache()
Manages cache directories for log files.
__init__
def __init__(cache_dir: Path = None)
Initialize cache manager.
Arguments:
cache_dir- Cache directory (defaults to CACHE_DIR)
get_dir
def get_dir(path: Path) -> Path
Get cache directory for a log file.
Arguments:
path- Path to the log file
Returns:
Cache directory path
Raises:
OSError- If file cannot be accessed or cache directory cannot be created
cleanup
def cleanup()
Clean up cache directories for files that no longer exist.
logloglog.log_file
Simple file abstraction for log file operations.
LogFile Objects
class LogFile()
Simple file abstraction for reading and writing log files.
Keeps file handle open during batch operations for performance. Call open() to start a batch read session, close() when done. Individual operations (append, get_size) open/close as needed.
__init__
def __init__(path: Union[Path, str], mode: str = "r")
Initialize LogFile.
Arguments:
path- Path to the log filemode- File mode - “r” for read-only, “a” for append, “w” for write
open
def open()
Open the file for reading. Call this before batch read operations.
close
def close()
Close the file handle. Call this after batch operations complete.
read_line
def read_line() -> Optional[str]
Read the next line from the current position.
Returns:
The next line without trailing newline, or None if no more data.
read_all_lines
def read_all_lines() -> list[str]
Read all remaining lines from current position.
Returns:
List of lines without trailing newlines.
append_line
def append_line(line: str) -> None
Append a line to the file.
Arguments:
line- Line to append (newline will be added automatically)
Raises:
IOError- If file is opened in read-only mode
append_lines
def append_lines(lines: list[str]) -> None
Append multiple lines to the file.
Arguments:
lines- Lines to append (newlines will be added as needed)
has_more_data
def has_more_data() -> bool
Check if there’s more data available to read.
Returns:
True if file has grown beyond current read position.
get_size
def get_size() -> int
Get current file size in bytes.
Returns:
File size in bytes, or 0 if file doesn’t exist.
seek_to
def seek_to(position: int) -> None
Set the read position.
Arguments:
position- Byte position to seek to
get_position
def get_position() -> int
Get current read position.
Returns:
Current byte position for reads.
reset
def reset() -> None
Reset read position to beginning of file.
aread_line
async def aread_line() -> Optional[str]
Async version of read_line().
aread_all_lines
async def aread_all_lines() -> list[str]
Async version of read_all_lines().
aappend_line
async def aappend_line(line: str) -> None
Async version of append_line().
aappend_lines
async def aappend_lines(lines: list[str]) -> None
Async version of append_lines().
ahas_more_data
async def ahas_more_data() -> bool
Async version of has_more_data().
aget_size
async def aget_size() -> int
Async version of get_size().
logloglog.logloglog
Main LogLogLog implementation.
default_get_width
@lru_cache(maxsize=100000)
def default_get_width(line: str) -> int
Fast line width calculation with ASCII fast path and caching.
default_split_lines
def default_split_lines(text: str) -> List[str]
Default line splitting on newlines.
LogLogLog Objects
class LogLogLog()
Efficient scrollback indexing for large log files.
LogLogLog provides O(log n) seeking through large logs at any terminal width.
__init__
def __init__(path: Union[Path, str, LogFile],
get_width: Callable[[str], int] = None,
split_lines: Callable[[str], List[str]] = None,
cache: Cache = None,
defer_indexing: bool = False)
Initialize LogLogLog for a file.
Arguments:
path- Log file path or LogFile instanceget_width- Function to calculate display width (defaults to wcwidth)split_lines- Function to split text into lines (defaults to newline split)cache- Cache instance (auto-created if None)defer_indexing- If True, skip initial indexing (useful for UI responsiveness)
aopen
async def aopen()
Async version of _open() method for non-blocking file initialization.
close
def close()
Close all resources.
__enter__
def __enter__()
Context manager entry.
__exit__
def __exit__(exc_type, exc_val, exc_tb)
Context manager exit.
update
def update()
Update index with new lines from the file.
aupdate
async def aupdate(progress_callback=None, progress_interval=0.1)
Async version of update() method for non-blocking file processing.
Arguments:
progress_callback- Optional async callback called periodically during indexingprogress_interval- Seconds between progress callbacks (default 0.1)
append
def append(line: str)
Append a line to the log file and update index.
Arguments:
line- Line to append (newline will be added)
Raises:
IOError- If LogFile was opened in read-only mode
__getitem__
def __getitem__(line_no: int) -> str
Get a logical line by line number.
__len__
def __len__() -> int
Get total number of logical lines.
__iter__
def __iter__() -> Iterator[str]
Iterate over all logical lines.
width
def width(width: int) -> WidthView
Create a width-specific view of this log.
Arguments:
width- Terminal width for line wrapping
Returns:
WidthView instance for this width
get_file_info
def get_file_info() -> dict
Get information about the log file.
Returns:
Dict with file size, current position, and other metadata.
get_cache_info
def get_cache_info() -> dict
Get information about the cache state.
Returns:
Dict with cache directory and status information.
line_at_row
def line_at_row(row: int, width: int) -> Tuple[int, int]
Find which logical line contains the given display row.
Arguments:
row- Display row numberwidth- Terminal width
Returns:
Tuple of (line_number, offset_within_line)
row_for_line
def row_for_line(line_no: int, width: int) -> int
Get the display row where a logical line starts.
Arguments:
line_no- Logical line numberwidth- Terminal width
Returns:
Display row number
total_rows
def total_rows(width: int) -> int
Get total number of display rows at given width.
Arguments:
width- Terminal width
Returns:
Total display rows