lsoph

lsoph.ui

lsoph.ui.file_data_table

A specialized DataTable widget using index-as-key for partial updates. NOTE: This approach uses the visual index as the RowKey, which can lead to unexpected behavior if rows are added/removed in ways that disrupt the expected visual order compared to the underlying data sort order.

COLUMN_KEYS

Order must match TableRow

FileDataTable Objects

class FileDataTable(DataTable)

A DataTable specialized for displaying and managing FileInfo. Uses stringified index as RowKey and attempts partial updates. Simplified error handling and logging. Attempts to maintain relative cursor screen position during updates.

RECENT_COL_WIDTH

Width for emoji history (e.g., 5 emojis + padding)

AGE_COL_WIDTH

width of the age column

SCROLLBAR_WIDTH

just a guess 🤷

COLUMN_PADDING

User’s estimate for padding per column

on_mount

def on_mount() -> None

Set up columns on mount.

selected_path

@property
def selected_path() -> Optional[str]

Returns the path string of the data visually at the cursor row.

on_resize

def on_resize(event: events.Resize) -> None

Update path column width on resize.

update_data

def update_data(infos: list[FileInfo]) -> None

Updates the table content using index as key and attempting partial updates. Attempts to maintain relative cursor screen position.

lsoph.ui.app

Main Textual application class for lsoph.

LsophApp Objects

class LsophApp(App[None])

Textual file monitor application for lsoph.

compose

def compose() -> ComposeResult

Create child widgets for the main application screen.

start_backend_worker

def start_backend_worker()

Starts the background worker to run the backend’s async method.

cancel_backend_worker

async def cancel_backend_worker()

Signals the backend instance to stop and cancels the Textual worker.

on_mount

def on_mount() -> None

Called when the app screen is mounted.

on_unmount

async def on_unmount() -> None

Called when the app is unmounted (e.g., on quit).

watch_last_monitor_version

def watch_last_monitor_version(old_version: int, new_version: int) -> None

Triggers table update when monitor version changes.

watch_status_text

def watch_status_text(old_text: str, new_text: str) -> None

Updates the status bar widget when status_text changes.

check_monitor_version

def check_monitor_version()

Periodically checks the monitor’s version and worker status.

update_status

def update_status(text: str)

Helper method to update the reactive status_text variable.

on_data_table_row_selected

def on_data_table_row_selected(event: DataTable.RowSelected) -> None

Handle row selection (Enter key) - show details.

on_data_table_row_activated

def on_data_table_row_activated(event: "DataTable.RowActivated") -> None

Handle row activation (Double Click) - show details.

action_quit

async def action_quit() -> None

Action to quit the application.

action_ignore_selected

def action_ignore_selected() -> None

Action to ignore the currently selected file path.

action_ignore_all

def action_ignore_all() -> None

Action to ignore all currently tracked files.

action_show_log

def action_show_log() -> None

Action to show or hide the log screen.

action_show_detail

def action_show_detail() -> None

Shows the detail screen for the selected row (requires focus check).

action_scroll_home

def action_scroll_home() -> None

Scrolls the file table to the top.

action_scroll_end

def action_scroll_end() -> None

Scrolls the file table to the bottom.

action_dump_monitor

def action_dump_monitor() -> None

Debug action to dump monitor state to log.

lsoph.ui.emoji

Generates emoji history strings for file activity.

get_emoji_history_string

def get_emoji_history_string(file_info: FileInfo, max_len: int = 5) -> str

Generates a string of emojis representing recent file activity history.

Arguments:

file_info - The FileInfo object containing the event history.
max_len - The maximum number of emojis to include in the string.

Returns:

A string of emojis (most recent first), padded with spaces.

lsoph.ui.detail_screen

Screen to display file event history using a DataTable.

DetailScreen Objects

class DetailScreen(Screen)

Screen to display event history and details for a specific file using DataTable.

compose

def compose() -> ComposeResult

Create child widgets for the detail screen.

on_mount

def on_mount() -> None

Called when the screen is mounted. Populates the DataTable.

lsoph.ui.log_screen

Full-screen display for application logs.

LogScreen Objects

class LogScreen(Screen)

A full screen to display application logs using RichLog.

compose

def compose() -> ComposeResult

Create child widgets for the log screen.

on_mount

def on_mount() -> None

Called when the screen is mounted. Populates with existing logs and starts timer.

on_unmount

def on_unmount() -> None

Called when the screen is unmounted. Stops the timer.

action_clear_log

def action_clear_log() -> None

Action to clear the log display.

lsoph.backend

LSOPH Backend Package.

log

Logger for this package

lsoph.backend.base

Base definitions for asynchronous monitoring backends.

Backend Objects

class Backend(ABC)

Abstract Base Class for all monitoring backends.

backend_name

Default, should be overridden

init

def __init__(monitor: Monitor)

Initialize the backend.

is_available

@staticmethod
@abstractmethod
def is_available() -> bool

Check if the backend’s dependencies (e.g., executable) are met. This MUST be implemented by subclasses.

attach

@abstractmethod
async def attach(pids: list[int])

Asynchronously attach to and monitor existing process IDs. Should periodically check self.should_stop.

run_command

async def run_command(command: list[str])

Default implementation to run a command and monitor it using the backend’s attach method. Backends like strace should override this if they have a different run mechanism.

stop

async def stop()

Signals the backend’s running task to stop and terminates the managed process if any.

should_stop

@property
def should_stop() -> bool

Check if the stop event has been set.

lsoph.backend.strace.handlers

Syscall handlers and CWD update logic for the strace backend.

update_cwd

def update_cwd(pid: int, cwd_map: dict[int, str], monitor: Monitor,
               event: Syscall)

Updates the CWD map based on chdir or fchdir syscalls.

lsoph.backend.strace.terminate

Contains logic for terminating the strace process.

log

Use module-specific logger

terminate_strace_process

async def terminate_strace_process(process: asyncio.subprocess.Process | None,
                                   pid: int)

Helper to terminate the strace process robustly.

lsoph.backend.strace

lsoph.backend.strace.backend

Strace backend implementation using refactored components.

Strace Objects

class Strace(Backend)

Async backend implementation using strace (refactored).

is_available

@staticmethod
def is_available() -> bool

Check if the strace executable is available in the system PATH.

attach

async def attach(pids: list[int])

Implementation of the attach method.

run_command

async def run_command(command: list[str])

Implementation of the run_command method.

stop

async def stop()

Signals the backend’s running task to stop and terminates the managed strace process.

lsoph.backend.strace.parse

Parsing logic for strace output.

Syscall Objects

@dataclass
class Syscall()

Represents a parsed strace syscall event.

timestamp

Timestamp when processed

raw_line

Store original line for debugging

parse_strace_stream

async def parse_strace_stream(
        lines: AsyncIterator[str],
        monitor: Monitor,
        stop_event: asyncio.Event,
        syscalls: list[str] | None = None,
        attach_ids: list[int] | None = None) -> AsyncIterator[Syscall]

Asynchronously parses a stream of raw strace output lines into Syscall objects.

lsoph.backend.strace.helpers

General helper functions for the strace backend.

parse_result_int

def parse_result_int(result_str: str) -> int | None

Safely parses an integer result string from strace.

clean_path_arg

def clean_path_arg(path_arg: Any) -> str | None

Cleans and decodes path arguments from strace, handling quotes and escapes.

parse_dirfd

def parse_dirfd(dirfd_arg: str | None) -> int | str | None

Parses the dirfd argument, handling AT_FDCWD.

resolve_path

def resolve_path(pid: int,
                 path: str | None,
                 cwd_map: dict[int, str],
                 monitor: Monitor,
                 dirfd: int | str | None = None) -> str | None

Resolves a path argument relative to CWD or dirfd if necessary.

lsoph.backend.lsof

Lsof backend package for lsoph.

lsoph.backend.lsof.backend

Lsof backend implementation using polling and descendant tracking.

log

Use specific logger

Lsof Objects

class Lsof(Backend)

Async backend implementation using periodic lsof command execution.

Monitors specified initial PIDs and automatically discovers and monitors their descendants over time. Detects file open/close events by comparing lsof output between poll cycles.

init

def __init__(
        monitor: Monitor,
        poll_interval: float = DEFAULT_LSOF_POLL_INTERVAL,
        child_check_multiplier: int = DEFAULT_CHILD_CHECK_INTERVAL_MULTIPLIER)

Initializes the Lsof backend.

Arguments:

monitor - The central Monitor instance.
poll_interval - Time in seconds between lsof polls.
child_check_multiplier - Check for new descendants every N polls.

is_available

@staticmethod
def is_available() -> bool

Check if the lsof executable is available in the system PATH.

attach

async def attach(pids: list[int])

Attaches to and monitors a list of initial PIDs and their descendants.

This method runs a continuous loop, polling with lsof, updating the monitor state, and periodically checking for new child processes of the initial PIDs.

Arguments:

pids - A list of initial process IDs to monitor.

lsoph.backend.lsof.parse

Parsing functions for lsof -F output.

log

Use specific logger

lsoph.backend.lsof.helpers

Helper functions for the lsof backend.

log

Use specific logger

lsoph.backend.psutil

Psutil backend package for lsoph.

lsoph.backend.psutil.backend

Psutil backend implementation using polling.

log

Use specific logger

Psutil Objects

class Psutil(Backend)

Async backend implementation using psutil polling.

is_available

@staticmethod
def is_available() -> bool

Check if the psutil library is installed.

attach

async def attach(pids: list[int])

Implementation of the attach method.

lsoph.backend.psutil.helpers

Helper functions for the psutil backend.

log

Use specific logger

lsoph.log

Logging setup for the lsoph application.

LOG_QUEUE

Max 1000 lines in memory

TextualLogHandler Objects

class TextualLogHandler(logging.Handler)

A logging handler that puts formatted messages into a deque for Textual.

emit

def emit(record: logging.LogRecord)

Formats the log record and adds it to the queue with Rich markup.

setup_logging

def setup_logging(level_name: str = "INFO")

Configures the root logger to use the TextualLogHandler.

lsoph.cli

parse_arguments

def parse_arguments(available_backends: dict[str, Type[Backend]],
                    argv: list[str] | None = None) -> argparse.Namespace

Parses command-line arguments for lsoph.

main

def main(argv: list[str] | None = None) -> int

Main entry point: Parses args, sets up logging, creates Monitor, instantiates backend, creates the specific backend coroutine, and launches the Textual UI.

lsoph.util

lsoph.util.pid

get_descendants

def get_descendants(parent_pid: int) -> list[int]

Retrieves a list of all descendant process IDs (PIDs) for a given parent PID.

get_cwd

def get_cwd(pid: int) -> str | None

Retrieves the Current Working Directory (CWD) for a given PID.

Uses psutil for cross-platform compatibility where possible, falling back to /proc//cwd on Linux if needed (though psutil usually handles this).

Arguments:

pid - The Process ID.

Returns:

The absolute path string of the CWD, or None if the process doesn’t exist, access is denied, or the CWD cannot be determined.

main

def main(argv: list[str] | None = None) -> int

Command-line entry point for testing pid functions.

lsoph.util.versioned

Provides a base class and decorators for simple version tracking and thread-safe access using locks. Useful for UI updates based on state changes.

Versioned Objects

class Versioned()

Base class for objects whose state changes should be trackable via a version number. Includes a reentrant lock for thread safety when modifying or accessing state.

init

def __init__()

Initializes version to 0 and creates a reentrant lock.

change

def change()

Manually increments the version number. Acquires the lock to ensure atomic update.

version

@property
def version() -> int

Returns the current version number of this object. Acquires the lock for thread-safe read.

hash

def __hash__() -> int

Makes Versioned objects hashable based on identity and current version. Useful for caching mechanisms that depend on object state. Acquires the lock for thread-safe read of version.

changes

def changes(method)

Decorator for methods that modify the state of a Versioned object. Acquires the object’s lock, executes the method, and then increments the version number atomically.

waits

def waits(method)

Decorator for methods that access the state of a Versioned object but do not modify it. Acquires the object’s lock before executing the method to ensure thread-safe reads, especially if accessing multiple attributes.

lsoph.util.short_path

Utility function for shortening file paths.

short_path

def short_path(path: str | os.PathLike,
               max_length: int,
               cwd: str = CWD) -> str

Shortens a file path string to fit max_length:

Tries to make path relative to CWD.
Prioritizes keeping the full filename visible.
If filename alone is too long, truncates filename from the left (“…name”).
If path is still too long but filename fits, truncates directory in the middle (“dir…ory/name”).

Arguments:

path - The file path string or path-like object.
max_length - The maximum allowed length for the output string.
cwd - The current working directory to make the path relative to.

Returns:

The shortened path string.

lsoph.monitor._fileinfo

Dataclass definition for storing information about a single tracked file.

FileInfo Objects

@dataclass
class FileInfo()

Holds state information about a single tracked file.

status

e.g., unknown, open, closed, active, deleted, error

last_event_type

e.g., OPEN, CLOSE, READ, WRITE, STAT, DELETE, RENAME

last_error_enoent

True if last relevant op failed with ENOENT

is_open

@property
def is_open() -> bool

Checks if any process currently holds this file open according to state.

lsoph.monitor

lsoph Monitor Package.

This package provides the core state management for monitored file access.

lsoph.monitor._monitor

Contains the Monitor class responsible for managing file access state.

log

Use the same logger name

Monitor Objects

class Monitor(Versioned)

Manages the state of files accessed by a monitored target (process group). Inherits from Versioned for change tracking and thread safety via decorators. Provides methods for backends to report file events (open, close, read, etc.).

ignore

@changes
def ignore(path: str)

Adds a path to the ignore list and removes existing state for it.

ignore_all

@changes
def ignore_all()

Adds all currently tracked file paths to the ignore list.

open

@changes
def open(pid: int, path: str, fd: int, success: bool, timestamp: float,
         **details)

Handles an ‘open’ or ‘creat’ event.

close

@changes
def close(pid: int, fd: int, success: bool, timestamp: float, **details)

Handles a ‘close’ event.

read

@changes
def read(pid: int, fd: int, path: str | None, success: bool, timestamp: float,
         **details)

Handles a ‘read’ (or similar) event.

write

@changes
def write(pid: int, fd: int, path: str | None, success: bool, timestamp: float,
          **details)

Handles a ‘write’ (or similar) event.

stat

@changes
def stat(pid: int, path: str, success: bool, timestamp: float, **details)

Handles a ‘stat’, ‘access’, ‘lstat’ etc. event.

delete

@changes
def delete(pid: int, path: str, success: bool, timestamp: float, **details)

Handles an ‘unlink’, ‘rmdir’ event.

rename

@changes
def rename(pid: int, old_path: str, new_path: str, success: bool,
           timestamp: float, **details)

Handles a ‘rename’ event.

process_exit

@changes
def process_exit(pid: int, timestamp: float)

Handles cleanup when a process exits.

get_path

@waits
def get_path(pid: int, fd: int) -> str | None

Retrieves the path for a PID/FD, handling standard streams.