Alternative Process Control

class PopenSpawn(SpawnBase):
    """
    Spawn class using subprocess.Popen for process communication.
    
    Uses pipes instead of pseudo-terminals, making it compatible with Windows
    and useful for scenarios where PTY behavior is not desired.
    """
    
    def __init__(self, cmd, timeout=30, maxread=2000, searchwindowsize=None,
                 logfile=None, cwd=None, env=None, encoding=None,
                 codec_errors='strict', preexec_fn=None):
        """
        Initialize PopenSpawn instance.
        
        Parameters:
        - cmd (str or list): Command to execute
        - timeout (int): Default timeout for operations
        - maxread (int): Maximum bytes to read at once
        - searchwindowsize (int): Search window size for pattern matching
        - logfile (file-like): File object for logging
        - cwd (str): Working directory for child process
        - env (dict): Environment variables for child process
        - encoding (str): Text encoding for I/O operations
        - codec_errors (str): How to handle encoding errors
        - preexec_fn (callable): Function to call before exec (Unix only)
        """

class fdspawn(SpawnBase):
    """
    Spawn-like interface for existing file descriptors.
    
    Allows expect functionality with sockets, named pipes (FIFOs),
    or any existing file descriptor.
    
    Note: On Windows, socket.fileno() doesn't provide a readable 
    file descriptor. Use SocketSpawn for cross-platform socket support.
    """
    
    def __init__(self, fd, args=None, timeout=30, maxread=2000,
                 searchwindowsize=None, logfile=None, encoding=None,
                 codec_errors='strict', use_poll=False):
        """
        Initialize fdspawn with existing file descriptor.
        
        Parameters:
        - fd (int): Existing file descriptor
        - args (list): Arguments for compatibility (not used)
        - timeout (int): Default timeout for operations
        - maxread (int): Maximum bytes to read at once
        - searchwindowsize (int): Search window size
        - logfile (file-like): File object for logging
        - encoding (str): Text encoding for I/O operations
        - codec_errors (str): How to handle encoding errors
        - use_poll (bool): Use poll() instead of select() for I/O
        """
    
    def isalive(self):
        """
        Check if file descriptor is still valid.
        
        Returns:
        bool: True if file descriptor is open and valid
        """
    
    def close(self):
        """Close the file descriptor."""
    
    def fileno(self):
        """
        Get the file descriptor number.
        
        Returns:
        int: File descriptor number
        """

class SocketSpawn(SpawnBase):
    """
    Spawn-like interface for socket communication.
    
    Provides expect functionality for socket connections,
    useful for network protocol automation and testing.
    """
    
    def __init__(self, socket, args=None, timeout=30, maxread=2000,
                 searchwindowsize=None, logfile=None, encoding=None,
                 codec_errors='strict'):
        """
        Initialize SocketSpawn with existing socket.
        
        Parameters:
        - socket (socket.socket): Connected socket object
        - args (list): Arguments for compatibility (not used)
        - timeout (int): Default timeout for operations
        - maxread (int): Maximum bytes to read at once
        - searchwindowsize (int): Search window size
        - logfile (file-like): File object for logging
        - encoding (str): Text encoding for I/O operations
        - codec_errors (str): How to handle encoding errors
        """
    
    def close(self):
        """Close the socket connection."""
    
    def isalive(self):
        """
        Check if socket connection is still active.
        
        Returns:
        bool: True if socket is connected
        """
    
    def fileno(self):
        """
        Get the socket file descriptor.
        
        Returns:
        int: Socket file descriptor
        """

from pexpect.popen_spawn import PopenSpawn
import sys

# PopenSpawn works on both Unix and Windows
if sys.platform == 'win32':
    # Windows command
    child = PopenSpawn('cmd.exe')
    child.expect('>')
    child.sendline('dir')
    child.expect('>')
    print(child.before.decode())
else:
    # Unix command
    child = PopenSpawn('bash')
    child.expect('$ ')
    child.sendline('ls -la')
    child.expect('$ ')
    print(child.before.decode())

child.close()

from pexpect.popen_spawn import PopenSpawn

# Execute complex command with arguments
child = PopenSpawn(['python', '-c', 'print("Hello"); input("Enter: ")'])

# Wait for the input prompt
child.expect('Enter: ')

# Send input
child.sendline('World')

# Read the output
child.expect(pexpect.EOF)
output = child.before.decode()
print(f"Output: {output}")

child.close()

from pexpect.fdpexpect import fdspawn
import os

# Create a named pipe (FIFO)
pipe_path = '/tmp/test_pipe'
if not os.path.exists(pipe_path):
    os.mkfifo(pipe_path)

# Open the pipe for reading/writing
fd = os.open(pipe_path, os.O_RDWR)

# Create fdspawn instance
child = fdspawn(fd)

# In another process/thread, write to the pipe
# For this example, we'll simulate it
os.write(fd, b'Hello from pipe\n')

# Use expect functionality
child.expect('Hello')
print(f"Received: {child.after.decode()}")

child.close()
os.unlink(pipe_path)  # Clean up

from pexpect.socket_pexpect import SocketSpawn
import socket

# Create and connect socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('httpbin.org', 80))

# Create SocketSpawn instance
child = SocketSpawn(sock)

# Send HTTP request
child.send('GET /get HTTP/1.1\r\n')
child.send('Host: httpbin.org\r\n')
child.send('Connection: close\r\n')
child.send('\r\n')

# Wait for HTTP response
child.expect('HTTP/')
print("Got HTTP response header")

# Read response body
child.expect('200 OK')
child.expect('\r\n\r\n')  # End of headers

# Read JSON response
child.expect(pexpect.EOF)
response_body = child.before.decode()
print(f"Response: {response_body}")

child.close()

from pexpect.socket_pexpect import SocketSpawn
import socket

# Connect to FTP server
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('ftp.example.com', 21))

child = SocketSpawn(sock)

# Wait for FTP welcome message
child.expect('220')
print("Connected to FTP server")

# Send username
child.sendline('USER anonymous')
child.expect('331')  # User name okay, need password

# Send password
child.sendline('PASS guest@example.com')
child.expect('230')  # User logged in
print("Logged in successfully")

# List directory
child.sendline('LIST')
child.expect('150')  # File status okay, about to open data connection
child.expect('226')  # Closing data connection
directory_listing = child.before.decode()
print(f"Directory listing:\n{directory_listing}")

# Quit
child.sendline('QUIT')
child.expect('221')  # Service closing control connection
print("Disconnected from FTP server")

child.close()

from pexpect.socket_pexpect import SocketSpawn
import socket

# Connect to telnet server
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('towel.blinkenlights.nl', 23))  # Star Wars ASCII movie

child = SocketSpawn(sock, encoding='utf-8')

try:
    # Just watch the ASCII movie for a bit
    for i in range(10):  # Watch for 10 iterations
        child.expect('\x1b[H')  # ANSI clear screen sequence
        frame = child.before
        print(f"Frame {i+1}")
        # Could process or display the frame here
        
except KeyboardInterrupt:
    print("Stopping...")

child.close()

from pexpect.fdpexpect import fdspawn
import subprocess
import os

# Create a process pipeline
p1 = subprocess.Popen(['ls', '-la'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', 'pexpect'], stdin=p1.stdout, stdout=subprocess.PIPE)

# Use fdspawn to interact with the pipeline output
child = fdspawn(p2.stdout.fileno())

try:
    # Read output with expect
    child.expect(pexpect.EOF)
    result = child.before.decode()
    print(f"Filtered output:\n{result}")
    
except pexpect.EOF:
    # Expected when process completes
    pass

# Clean up
p1.stdout.close()
p2.stdout.close()
p1.wait()
p2.wait()
child.close()

from pexpect.popen_spawn import PopenSpawn
import sys

if sys.platform == 'win32':
    # Windows PowerShell automation
    child = PopenSpawn(['powershell.exe', '-Command', '-'])
    
    # Wait for PowerShell prompt
    child.expect('PS ')
    
    # Execute PowerShell commands
    child.sendline('Get-Date')
    child.expect('PS ')
    date_output = child.before.decode()
    print(f"Current date: {date_output}")
    
    child.sendline('Get-Process | Select-Object -First 5')
    child.expect('PS ')
    process_output = child.before.decode()
    print(f"Top processes:\n{process_output}")
    
    # Exit PowerShell
    child.sendline('exit')
    child.close()
    
else:
    print("This example is for Windows only")