tvmux

tvmux.proc.bg

Background process management with automatic cleanup.

spawn

def spawn(cmd: List[str], **kwargs) -> subprocess.Popen

Run a subprocess in the background with automatic cleanup on exit.

Arguments:

Returns:

The Popen process object

terminate

def terminate(pid: int) -> bool

Stop a tracked background process.

Arguments:

Returns:

True if process was stopped, False if not found

reap

def reap()

Remove PIDs of processes that have already exited.

tvmux.proc

Process utilities.

run

def run(cmd: List[str], **kwargs) -> subprocess.CompletedProcess

Run a subprocess synchronously with automatic logging.

Arguments:

Returns:

CompletedProcess result

run_bg

async def run_bg(cmd: List[str], **kwargs) -> subprocess.Popen

Run a subprocess in the background asynchronously.

Arguments:

Returns:

The Popen process object

tvmux.models.remote

Base model for objects that sync with a remote backend.

RemoteModel Objects

class RemoteModel(BaseModel)

Base model that tracks changes and syncs with remote backend.

__enter__

def __enter__()

Enter transaction mode - disable auto-sync.

__exit__

def __exit__(exc_type, exc_val, exc_tb)

Exit transaction mode - commit changes.

from_remote

@classmethod
def from_remote(cls, **data)

Create instance from remote data without syncing back.

dirty_fields

@property
def dirty_fields()

Get list of fields with pending changes.

commit

def commit()

Commit pending changes to remote backend.

Subclasses must implement this method.

tvmux.models.session

Session model for tvmux.

Session Objects

class Session(BaseModel)

A tmux session.

tvmux.models.position

Position type for dimensions.

Position Objects

class Position(BaseModel)

A position or size as (x, y) coordinates.

from_string

@classmethod
def from_string(cls, value: str) -> "Position"

Parse from tmux format like ‘80x24’.

__str__

def __str__() -> str

Format as tmux string.

as_tuple

def as_tuple() -> Tuple[int, int]

Get as tuple.

tvmux.models.recording

Recording model for tvmux.

Recording Objects

class Recording(BaseModel)

A tmux recording session.

start

async def start(active_pane: str, output_dir: Path)

Start recording this window.

switch_pane

def switch_pane(new_pane_id: str)

Switch recording to a different pane in the window.

stop

def stop()

Stop recording.

tvmux.models

Models for tvmux.

tvmux.models.window

Window model for tvmux.

Window Objects

class Window(BaseModel)

A tmux window.

tvmux.models.pane

Pane model for tvmux.

Pane Objects

class Pane(BaseModel)

A tmux pane.

tvmux.connection

Connection to tvmux server.

Connection Objects

class Connection()

Manages connection to tvmux server.

server_pid

@property
def server_pid() -> Optional[int]

Get server PID if running.

is_running

@property
def is_running() -> bool

Check if server is running.

api

def api()

Get API client with typed methods matching the server routes.

tvmux.repair

Asciinema cast file repair utilities.

Simple functions to detect and repair corrupted asciinema cast files. Handles large files by streaming instead of loading into RAM.

validate_cast_file

def validate_cast_file(cast_path: Path) -> bool

Validate that an asciinema cast file is properly formatted. Only reads header and tail to avoid loading large files into RAM.

Arguments:

Returns:

True if file is valid, False otherwise

repair_cast_file

def repair_cast_file(cast_path: Path, backup: bool = True) -> bool

Repair a corrupted asciinema cast file.

Streams the file to avoid memory issues with large files. Fixes incomplete JSON arrays due to asciinema termination during write.

Arguments:

Returns:

True if repair was successful or unnecessary, False on failure

tvmux.tui.app

Main TUI application with CRT TV interface.

setup_client_logging

def setup_client_logging()

Set up client-side logging.

ChannelTuner Objects

class ChannelTuner(Static)

TV channel tuner showing tmux windows as channels.

on_mount

async def on_mount() -> None

Load channels when widget mounts.

refresh_channels

async def refresh_channels() -> None

Refresh the list of available tmux windows/sessions.

render

def render() -> str

Render the channel tuner.

action_select_next

def action_select_next() -> None

Select next channel.

action_select_previous

def action_select_previous() -> None

Select previous channel.

get_selected_channel

def get_selected_channel() -> Optional[dict]

Get the currently selected channel.

toggle_recording

async def toggle_recording() -> None

Start or stop recording for the selected channel.

CRTPlayer Objects

class CRTPlayer(Static)

CRT-style video player widget.

compose

def compose() -> ComposeResult

Compose the CRT player.

play_recording

async def play_recording(recording_path: Path) -> None

Play a recording file.

show_blank

async def show_blank() -> None

Show a blank/static screen for channels not recording.

TVMuxApp Objects

class TVMuxApp(App)

Main tvmux TUI application.

compose

def compose() -> ComposeResult

Compose the application layout.

on_mount

async def on_mount() -> None

Initialize player when app starts.

action_refresh

async def action_refresh() -> None

Refresh channels list.

action_select_next

def action_select_next() -> None

Select next channel.

action_select_previous

def action_select_previous() -> None

Select previous channel.

action_play_selected

async def action_play_selected() -> None

Tune to the selected channel.

action_toggle_playback

async def action_toggle_playback() -> None

Toggle recording for current channel.

schedule_channel_check

def schedule_channel_check() -> None

Schedule a channel check for auto-play.

tune_to_selected_channel

async def tune_to_selected_channel() -> None

Auto-play the selected channel if it’s recording.

run_tui

def run_tui()

Run the tvmux TUI application.

tvmux.tui

TUI components for tvmux.

tvmux.api_client

Simple API client function for tvmux server.

APIError Objects

class APIError(Exception)

API request failed.

api_call

def api_call(base_url: str,
             method: str,
             path: str,
             data: Optional[BaseModel] = None,
             response_model: Optional[Type[T]] = None) -> Union[T, dict]

Make an API call with Pydantic support.

Arguments:

Returns:

Instance of response_model or dict

Raises:

tvmux.utils

Utility functions for tvmux.

get_session_dir

def get_session_dir(hostname: str,
                    session_name: str,
                    tmux_var: str,
                    base_dir: str = "/run/tvmux") -> Path

Generate a filesystem-safe session directory name.

Arguments:

Returns:

Path to the session directory

Example:

get_session_dir(“laptop”, “my project”, “/tmp/tmux-1000/default,3028,0”) PosixPath(‘/run/tvmux/session_laptop_my_project_a1b2c3’)

safe_filename

def safe_filename(name: str) -> str

Make a string safe for use as a filename.

file_has_readers

def file_has_readers(file_path: str) -> bool

Check if any process is set up to read from the file.

tvmux.config

Configuration management for tvmux.

OutputConfig Objects

class OutputConfig(BaseModel)

Output configuration.

ServerConfig Objects

class ServerConfig(BaseModel)

Server configuration.

RecordingConfig Objects

class RecordingConfig(BaseModel)

Recording configuration.

AnnotationConfig Objects

class AnnotationConfig(BaseModel)

Annotation configuration.

LoggingConfig Objects

class LoggingConfig(BaseModel)

Logging configuration.

Config Objects

class Config(BaseModel)

Main tvmux configuration.

load_config

def load_config(config_file: Optional[str] = None) -> Config

Load configuration from file and environment variables.

Precedence (highest to lowest):

  1. Environment variables
  2. Specified config file
  3. TVMUX_CONFIG_FILE environment variable
  4. ~/.tvmux.conf
  5. Built-in defaults

generate_env_var_name

def generate_env_var_name(section: str, field: str) -> str

Generate environment variable name for a config field.

Follows convention: TVMUX_{SECTION}_{FIELD}

get_all_env_mappings

def get_all_env_mappings() -> Dict[str, tuple[str, str]]

Get all possible environment variable mappings.

Returns dict mapping env var name to (section, field) tuple.

load_all_env_overrides

def load_all_env_overrides() -> Dict[str, Any]

Load all environment variable overrides programmatically.

dump_config_toml

def dump_config_toml(config: Config) -> str

Convert Config to TOML string.

dump_config_env

def dump_config_env(config: Config) -> str

Convert Config to environment variable format.

get_config

def get_config() -> Config

Get the global configuration instance.

set_config

def set_config(config: Config) -> None

Set the global configuration instance.

tvmux.cli.main

Main CLI entry point for tvmux.

def print_version(ctx, param, value)

Print version information for client and server.

cli

@click.group(invoke_without_command=True)
@click.option('--log-level',
              default='INFO',
              type=click.Choice(['DEBUG', 'INFO', 'WARNING', 'ERROR']),
              help='Set logging level')
@click.option('--config-file',
              type=click.Path(exists=True),
              help='Path to configuration file')
@click.option('--version',
              is_flag=True,
              callback=print_version,
              expose_value=False,
              is_eager=True,
              help='Show version information')
@click.pass_context
def cli(ctx, log_level, config_file)

Per-window recorder for tmux.

tvmux.cli.record

Recording management commands.

rec

@click.group(invoke_without_command=True)
@click.pass_context
def rec(ctx)

Manage window recordings.

start

@rec.command("start")
def start()

Start recording the current tmux window.

ls

@rec.command("ls")
@click.option("-q",
              "--quiet",
              is_flag=True,
              help="Only output recording IDs (one per line)")
def ls(quiet)

List active recordings.

stop

@rec.command("stop")
@click.argument("recording_ids", nargs=-1)
def stop(recording_ids)

Stop recording(s). Stop all recordings if no IDs specified.

tvmux.cli.api_cli

Better auto-generated CLI from FastAPI routes using introspection.

pydantic_to_click_options

def pydantic_to_click_options(model: type[BaseModel])

Convert Pydantic model fields to Click options.

create_command_for_route

def create_command_for_route(route: APIRoute)

Create a Click command for a FastAPI route.

api

@click.group()
def api()

Direct API access for testing (auto-generated from routes).

generate_cli

def generate_cli()

Generate CLI commands from FastAPI routes.

tvmux.cli

CLI module for tvmux.

tvmux.cli.server

Server management commands.

server

@click.group()
def server()

Manage the tvmux server.

tvmux.cli.config

Configuration management commands.

config

@click.group()
def config()

Configuration management commands.

show

@config.command()
@click.option('--format',
              'output_format',
              default='toml',
              type=click.Choice(['toml', 'env']),
              help='Output format (toml or env)')
def show(output_format)

Show current effective configuration.

defaults

@config.command()
@click.option('--format',
              'output_format',
              default='toml',
              type=click.Choice(['toml', 'env']),
              help='Output format (toml or env)')
def defaults(output_format)

Show default configuration values.

tvmux.cli.tui

TUI command for tvmux.

tui

@click.command()
def tui()

Launch the tvmux TUI interface.

tvmux.server.main

FastAPI server that manages tmux connections.

setup_logging

def setup_logging()

Configure logging for the application.

lifespan

@asynccontextmanager
async def lifespan(app: FastAPI)

Manage application lifespan.

root

@app.get("/")
async def root()

Server info.

version

@app.get("/version")
async def version()

Get server version.

cleanup_and_exit

def cleanup_and_exit(signum=None, frame=None)

Clean up and exit gracefully.

run_server

def run_server()

Run the server on HTTP port.

tvmux.server.state

Global state management for tvmux server.

SERVER_PORT

“TV” in ASCII

tvmux.server.window_monitor

Monitor tmux windows to detect when they’re closed.

get_current_windows

def get_current_windows() -> Set[str]

Get the set of current window IDs as session:window_id keys.

cleanup_closed_windows

def cleanup_closed_windows()

Check for closed windows and clean up their recordings.

tvmux.server

tvmux server package.

tvmux.server.routers.session

Session router for tmux control.

SessionCreate Objects

class SessionCreate(BaseModel)

Create session request.

SessionUpdate Objects

class SessionUpdate(BaseModel)

Update session request.

WindowReference Objects

class WindowReference(BaseModel)

Reference to a window in a session.

SessionWindows Objects

class SessionWindows(BaseModel)

Session windows response.

update

@router.patch("/{session_id}")
async def update(session_id: str, update: SessionUpdate)

Update a session (rename).

attach_session

@router.post("/{session_id}/attach")
async def attach_session(session_id: str)

Attach to a session (returns attach command for client to execute).

detach_session

@router.post("/{session_id}/detach")
async def detach_session(session_id: str)

Detach all clients from a session.

get_session_windows

@router.get("/{session_id}/windows", response_model=SessionWindows)
async def get_session_windows(session_id: str)

Get all window references for a session.

tvmux.server.routers.panes

Pane router for tmux control.

PaneCreate Objects

class PaneCreate(BaseModel)

Create pane request.

window_id

Window to split

target_pane_id

Specific pane to split (default: active)

horizontal

False for vertical split, True for horizontal

size

Percentage or lines/columns

PaneResize Objects

class PaneResize(BaseModel)

Resize pane request.

direction

U, D, L, or R

PaneSendKeys Objects

class PaneSendKeys(BaseModel)

Send keys request.

list_panes

@router.get("/", response_model=List[Pane])
async def list_panes(window_id: Optional[str] = Query(
    None, description="Filter by window ID"))

List all panes or panes in a specific window.

create_pane

@router.post("/", response_model=Pane)
async def create_pane(pane: PaneCreate)

Create a new pane by splitting a window.

get_pane

@router.get("/{pane_id}", response_model=Pane)
async def get_pane(pane_id: str)

Get a specific pane by ID.

delete_pane

@router.delete("/{pane_id}")
async def delete_pane(pane_id: str)

Kill a pane.

select_pane

@router.post("/{pane_id}/select")
async def select_pane(pane_id: str)

Select/switch to a pane.

resize_pane

@router.post("/{pane_id}/resize")
async def resize_pane(pane_id: str, resize: PaneResize)

Resize a pane.

send_keys

@router.post("/{pane_id}/send-keys")
async def send_keys(pane_id: str, send: PaneSendKeys)

Send keys to a pane.

capture_pane

@router.get("/{pane_id}/capture")
async def capture_pane(pane_id: str,
                       start: Optional[int] = Query(None),
                       end: Optional[int] = Query(None))

Capture pane contents.

tvmux.server.routers.recording

Recording management endpoints.

resolve_id

def resolve_id(session_id: str, window_name: str) -> str

Get window ID from window name/index/id.

Arguments:

Returns:

Window ID (e.g., “@1”)

display_name

def display_name(session_id: str, window_id: str) -> str

Get friendly display name for a window ID.

RecordingCreate Objects

class RecordingCreate(BaseModel)

Request to start recording a window.

window_id

Window ID to record

active_pane

If not provided, will detect active pane

create_recording

@router.post("/", response_model=Recording)
async def create_recording(request: RecordingCreate,
                           response: Response) -> Recording

Start a new recording.

delete_recording

@router.delete("/{recording_id}")
async def delete_recording(recording_id: str) -> dict

Stop a recording.

get_recording

@router.get("/{recording_id}", response_model=Recording)
async def get_recording(recording_id: str) -> Recording

Get recording status.

list_recordings

@router.get("/", response_model=list[Recording])
async def list_recordings() -> list[Recording]

List all active recordings.

tvmux.server.routers

tvmux server routers.

tvmux.server.routers.hook

Single endpoint for receiving tmux hook events.

HookEvent Objects

class HookEvent(BaseModel)

Event data from tmux hooks.

receive_hook

@router.post("/")
async def receive_hook(event: HookEvent) -> Dict[str, str]

Receive and process a hook event from tmux.

tvmux.server.routers.window

Window router for tmux control.

WindowCreate Objects

class WindowCreate(BaseModel)

Create window request.

session

Session to attach to (optional)

WindowUpdate Objects

class WindowUpdate(BaseModel)

Update window request.

update_window

@router.patch("/{window_id}")
async def update_window(window_id: str, update: WindowUpdate)

Update a window.

delete_window

@router.delete("/{window_id}")
async def delete_window(window_id: str)

Kill a tmux window.

select_window

@router.post("/{window_id}/select")
async def select_window(window_id: str)

Select/switch to a window.

@router.post("/{window_id}/unlink")
async def unlink_window(window_id: str)

Unlink window from its session.

@router.post("/{window_id}/link")
async def link_window(window_id: str,
                      target_session: str,
                      target_index: Optional[int] = None)

Link window to a session.

get_window_panes

@router.get("/{window_id}/panes")
async def get_window_panes(window_id: str)

Get all panes in a window - redirects to panes endpoint.

tvmux.server.routers.callbacks

CRUD endpoints for managing tmux hooks.

Hook Objects

class Hook(BaseModel)

Configuration for a tmux hook.

command

If None, use default command

HookCreate Objects

class HookCreate(BaseModel)

Request to create/install a hook.

HookUpdate Objects

class HookUpdate(BaseModel)

Request to update a hook.

get_default_command

def get_default_command(hook_name: str) -> str

Get the default command for a hook.

install_hook

def install_hook(hook: Hook) -> None

Install a tmux hook.

uninstall_hook

def uninstall_hook(hook_name: str) -> None

Uninstall a tmux hook.

list_hooks

@router.get("/")
async def list_hooks() -> List[Hook]

List all available hooks and their status.

get_hook

@router.get("/{hook_name}")
async def get_hook(hook_name: str) -> Hook

Get details about a specific hook.

create_hook

@router.post("/")
async def create_hook(hook_data: HookCreate) -> Hook

Create/install a new hook.

update_hook

@router.put("/{hook_name}")
async def update_hook(hook_name: str, update_data: HookUpdate) -> Hook

Update an existing hook.

delete_hook

@router.delete("/{hook_name}")
async def delete_hook(hook_name: str) -> Dict[str, str]

Remove/uninstall a hook.

setup_default_hooks

def setup_default_hooks()

Set up default tmux hooks for tvmux operation.

remove_all_hooks

def remove_all_hooks()

Remove all installed tmux hooks.