.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:
base_url
- Server base URLmethod
- HTTP methodpath
- API endpoint pathdata
- Request body as Pydantic modelresponse_model
- Expected response model class
Returns:
Instance of response_model or dict
Raises:
APIError
- If the request fails
.tvmux
Tells Python that this is a module dir
.tvmux.cli
CLI module for tvmux.
.tvmux.cli.main
Main CLI entry point for tvmux.
cli
@click.group()
@click.option('--log-level',
default='INFO',
type=click.Choice(['DEBUG', 'INFO', 'WARNING', 'ERROR']),
help='Set logging level')
@click.version_option()
def cli(log_level)
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")
def ls()
List active recordings.
stop
@rec.command("stop")
@click.argument("recording_id", required=False)
def stop(recording_id)
Stop recording(s). Stop all recordings if no ID specified.
.tvmux.cli.server
Server management commands.
server
@click.group()
def server()
Manage the tvmux server.
.tvmux.server
tvmux server package.
.tvmux.server.routers
tvmux server routers.
.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.
unlink_window
@router.post("/{window_id}/unlink")
async def unlink_window(window_id: str)
Unlink window from its session.
link_window
@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.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("/{name}")
async def update(name: str, update: SessionUpdate)
Update a session (rename).
attach_session
@router.post("/{name}/attach")
async def attach_session(name: str)
Attach to a session (returns attach command for client to execute).
detach_session
@router.post("/{name}/detach")
async def detach_session(name: str)
Detach all clients from a session.
get_session_windows
@router.get("/{name}/windows", response_model=SessionWindows)
async def get_session_windows(name: str)
Get all window references for a session.
.tvmux.server.routers.callback
Callback endpoints for tmux hooks.
CallbackEvent Objects
class CallbackEvent(BaseModel)
Event data from tmux hooks.
window_index
Changed to string to handle empty values
pane_index
Changed to string to handle empty values
CallbackEventResponse Objects
class CallbackEventResponse(BaseModel)
Response for callback events.
HookConfig Objects
class HookConfig(BaseModel)
Configuration for a tmux hook.
list_callbacks
@router.get("/")
async def list_callbacks() -> List[CallbackEventResponse]
List recent callback events.
create_callback
@router.post("/")
async def create_callback(event: CallbackEvent) -> CallbackEventResponse
Create a new callback event (called by tmux hooks).
get_callback
@router.get("/{event_id}")
async def get_callback(event_id: str) -> CallbackEventResponse
Get a specific callback event.
delete_callback
@router.delete("/{event_id}")
async def delete_callback(event_id: str) -> Dict[str, str]
Remove a callback event from history.
setup_tmux_hooks
def setup_tmux_hooks()
Set up tmux hooks to call our callbacks.
remove_tmux_hooks
def remove_tmux_hooks()
Remove our tmux hooks.
.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:
session_id
- The session IDwindow_name
- Window name, index, or ID
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
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.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.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.
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.models
Models for tvmux.
.tvmux.models.window
Window model for tvmux.
Window Objects
class Window(BaseModel)
A tmux window.
.tvmux.models.session
Session model for tvmux.
Session Objects
class Session(BaseModel)
A tmux session.
.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.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.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.pane
Pane model for tvmux.
Pane Objects
class Pane(BaseModel)
A tmux pane.
.tvmux.proc
Process utilities.
run
def run(cmd: List[str], **kwargs) -> subprocess.CompletedProcess
Run a subprocess synchronously with automatic logging.
Arguments:
cmd
- Command to run as list of strings**kwargs
- Additional arguments passed to subprocess.run
Returns:
CompletedProcess result
run_bg
async def run_bg(cmd: List[str], **kwargs) -> subprocess.Popen
Run a subprocess in the background asynchronously.
Arguments:
cmd
- Command to run as list of strings**kwargs
- Additional arguments passed to subprocess.Popen
Returns:
The Popen process object
.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:
cmd
- Command to run as list of strings**kwargs
- Additional arguments passed to subprocess.Popen
Returns:
The Popen process object
terminate
def terminate(pid: int) -> bool
Stop a tracked background process.
Arguments:
pid
- Process ID to stop
Returns:
True if process was stopped, False if not found
reap
def reap()
Remove PIDs of processes that have already exited.
.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:
hostname
- The hostname where tmux is runningsession_name
- The tmux session nametmux_var
- The $TMUX environment variable valuebase_dir
- Base directory for tvmux runtime data
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.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:
cast_path
- Path to the cast file
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:
cast_path
- Path to the cast file to repairbackup
- Whether to create a backup before repair
Returns:
True if repair was successful or unnecessary, False on failure