Core Concepts

from termitty import TermittySession

# Create a session - the foundation of all automation
with TermittySession() as session:
    # Connect with various auth methods
    session.connect('server.com', username='deploy', key_file='~/.ssh/id_rsa')
    
    # Session maintains full terminal state
    session.execute('cd /var/log')
    session.execute('ls -la')
    
    # AI can access complete terminal state at any time
    state = session.state.terminal.get_structured_state()
    print(state['context']['working_directory'])  # /var/log

from termitty.conditions import OutputContains, PromptReady, TextContains

# Wait for specific output - no more sleep()!
session.execute('./deploy.sh', wait=False)
session.wait_until(OutputContains('Deployment successful'), timeout=300)

# Wait for prompts AI can handle
session.execute('sudo systemctl restart nginx', wait=False)
if session.wait_until(TextContains('[sudo] password'), timeout=10):
    session.send_line(get_sudo_password(), secure=True)

# Wait for prompt to be ready for next command
session.wait_until(PromptReady())

# Custom conditions for AI logic
session.wait_until(
    lambda s: 'error' not in s.terminal.get_screen_text().lower(),
    timeout=60,
    check_interval=1
)

# AI gets complete terminal state as structured data
state = session.state.terminal.get_structured_state()

# Example state structure:
{
    "screen": {
        "text": "user@server:~/app$ git status\nOn branch main\nnothing to commit",
        "cursor": {"row": 2, "col": 0},
        "size": {"rows": 24, "cols": 80}
    },
    "detected_elements": [
        {
            "type": "prompt", 
            "text": "user@server:~/app$", 
            "position": {"row": 0, "col": 0}
        },
        {
            "type": "git_status",
            "branch": "main",
            "status": "clean"
        }
    ],
    "context": {
        "working_directory": "/home/user/app",
        "last_command": "git status",
        "last_exit_code": 0,
        "environment": {
            "USER": "user",
            "HOME": "/home/user",
            "PATH": "/usr/bin:/bin"
        }
    },
    "session_info": {
        "connected_since": "2024-01-01T10:00:00Z",
        "commands_executed": 15,
        "current_shell": "/bin/bash"
    }
}

from termitty.keys import Keys

# AI can control any interactive application
with session.interactive_shell() as shell:
    # Navigate an installer
    shell.send_line('./install.sh')
    shell.wait_for_text('1. Express Install')
    shell.send_line('1')
    
    # Handle configuration prompts
    shell.wait_for_text('Enter port:')
    shell.send_line('8080')
    
    # Edit files with vim/nano
    shell.send_line('vim config.yaml')
    shell.wait_for_text('~')  # Wait for vim to load
    shell.send_key('i')  # Insert mode
    shell.send_keys('server_port: 8080\n')
    shell.send_key(Keys.ESCAPE)
    shell.send_line(':wq')
    
    # Get current screen for AI analysis
    screen = shell.get_screen_text()
    cursor_pos = shell.get_cursor_position()
    
    # AI can understand what's happening
    if 'error' in screen.lower():
        # AI handles errors intelligently
        handle_error(screen)

from termitty.parallel import ConnectionPool

# AI manages fleets of servers
pool = ConnectionPool([
    'web-01.example.com',
    'web-02.example.com', 
    'web-03.example.com'
], username='deploy', key_file='~/.ssh/id_rsa')

# Rolling deployments with AI monitoring
results = pool.execute_on_all(
    command='./deploy.sh v2.1.0',
    strategy='rolling',  # One at a time
    batch_size=1,
    on_error=ai_error_handler  # AI decides what to do on errors
)

# AI analyzes results across all servers
for result in results:
    if result.exit_code != 0:
        ai_handle_deployment_failure(result.host, result.output)
    else:
        ai_verify_deployment_success(result.host)

# Record expert actions for AI training
session.start_recording('expert_debug_session.json')

# Expert performs complex debugging
session.execute('docker ps | grep failing')
session.execute('docker logs container_id')
# ... expert identifies and fixes issue ...

recording = session.stop_recording()

# Recording contains:
# - Every command and output
# - Terminal state at each step  
# - Timing information
# - Context and environment
# Perfect for training AI models!

# Replay for AI learning
session.replay_recording('expert_debug_session.json')
ai_model.learn_from_session(recording)

from termitty.exceptions import CommandTimeout, ConnectionError

# AI-friendly error handling
class AITerminalAgent:
    def execute_with_recovery(self, command, max_retries=3):
        for attempt in range(max_retries):
            try:
                result = self.session.execute(command, timeout=60)
                
                if result.exit_code == 0:
                    return result
                else:
                    # AI analyzes the error
                    error_analysis = self.ai.analyze_error(
                        command=command,
                        output=result.output,
                        terminal_state=self.session.state.terminal.get_structured_state()
                    )
                    
                    if error_analysis['can_fix']:
                        # AI suggests a fix
                        fix_command = error_analysis['fix_command']
                        self.session.execute(fix_command)
                        continue
                    else:
                        # Escalate to human or alternative strategy
                        return self.escalate_error(command, result)
                        
            except CommandTimeout:
                # AI decides: kill process, wait longer, or try alternative
                action = self.ai.handle_timeout(command)
                if action == 'kill':
                    self.session.send_keys('^C')  # Ctrl+C
                elif action == 'wait':
                    continue
                    
            except ConnectionError:
                # AI tries reconnection or alternative servers
                if not self.reconnect():
                    return self.try_alternative_server()
        
        return None  # All retries failed