from abc import ABC, abstractmethod
class ProcessHandler(ABC):
"""Base class for process-specific handlers."""
def __init__(self, multiplexer):
self.multiplexer = multiplexer
self.state_machine = {}
self.current_state = "initial"
self.prompt_patterns = []
self.response_suggestions = {}
@abstractmethod
def get_process_type(self):
"""Return the process type this handler manages."""
def update_state(self, output):
"""Update internal state based on output."""
def get_prompt_suggestions(self):
"""Return suggested responses for current state."""
return self.response_suggestions.get(self.current_state, [])
def is_waiting_for_input(self):
"""Check if process appears to be waiting for input."""
return self.current_state in ["waiting_confirmation", "waiting_input"]
class AptHandler(ProcessHandler):
"""Handler for apt package manager interactions."""
def __init__(self, multiplexer):
super().__init__(multiplexer)
self.state_machine = {
"initial": ["running_command"],
"running_command": ["waiting_confirmation", "completed"],
"waiting_confirmation": ["confirmed", "cancelled"],
"confirmed": ["installing", "completed"],
"installing": ["completed", "error"],
"completed": [],
"error": [],
"cancelled": [],
}
self.prompt_patterns = [
("Do you want to continue\\?", "confirmation"),
("After this operation.*installed\\.", "size_info"),
("Need to get.*B of archives\\.", "download_info"),
("Unpacking.*Configuring", "configuring"),
("Setting up", "setting_up"),
("E:\\s", "error"),
]
def get_process_type(self):
return "apt"
def update_state(self, output):
"""Update state based on apt output patterns."""
output_lower = output.lower()
if "processing triggers" in output_lower or "done" in output_lower:
self.current_state = "completed"
elif "do you want to continue" in output_lower:
self.current_state = "waiting_confirmation"
elif "setting up" in output_lower or "unpacking" in output_lower:
self.current_state = "installing"
elif "e:" in output_lower or "error" in output_lower:
self.current_state = "error"
def get_prompt_suggestions(self):
"""Return suggested responses for apt prompts."""
suggestions = super().get_prompt_suggestions()
if self.current_state == "waiting_confirmation":
suggestions.extend(["y", "yes", "n", "no"])
return suggestions
class VimHandler(ProcessHandler):
"""Handler for vim editor interactions."""
def __init__(self, multiplexer):
super().__init__(multiplexer)
self.state_machine = {
"initial": ["normal_mode", "insert_mode"],
"normal_mode": ["insert_mode", "command_mode", "visual_mode"],
"insert_mode": ["normal_mode"],
"command_mode": ["normal_mode"],
"visual_mode": ["normal_mode"],
"exiting": [],
}
self.prompt_patterns = [
("-- INSERT --", "insert_mode"),
("-- VISUAL --", "visual_mode"),
(":", "command_mode"),
("Press ENTER", "waiting_enter"),
("Saved", "saved"),
]
self.mode_indicators = {"insert": "-- INSERT --", "visual": "-- VISUAL --", "command": ":"}
def get_process_type(self):
return "vim"
def update_state(self, output):
"""Update state based on vim mode indicators."""
if "-- INSERT --" in output:
self.current_state = "insert_mode"
elif "-- VISUAL --" in output:
self.current_state = "visual_mode"
elif output.strip().endswith(":"):
self.current_state = "command_mode"
elif "Press ENTER" in output:
self.current_state = "waiting_enter"
else:
self.current_state = "normal_mode"
def get_prompt_suggestions(self):
"""Return suggested commands for vim modes."""
suggestions = super().get_prompt_suggestions()
if self.current_state == "command_mode":
suggestions.extend(["w", "q", "wq", "q!", "w!"])
elif self.current_state == "normal_mode":
suggestions.extend(["i", "a", "o", "dd", ":w", ":q"])
elif self.current_state == "waiting_enter":
suggestions.extend(["\n"])
return suggestions
class SSHHandler(ProcessHandler):
"""Handler for SSH connection interactions."""
def __init__(self, multiplexer):
super().__init__(multiplexer)
self.state_machine = {
"initial": ["connecting"],
"connecting": ["auth_prompt", "connected", "failed"],
"auth_prompt": ["connected", "failed"],
"connected": ["shell", "disconnected"],
"shell": ["disconnected"],
"failed": [],
"disconnected": [],
}
self.prompt_patterns = [
("password:", "password_prompt"),
("yes/no", "host_key_prompt"),
("Permission denied", "auth_failed"),
("Welcome to", "connected"),
("\\$", "shell_prompt"),
("\\#", "root_shell_prompt"),
("Connection closed", "disconnected"),
]
def get_process_type(self):
return "ssh"
def update_state(self, output):
"""Update state based on SSH connection output."""
output_lower = output.lower()
if "permission denied" in output_lower:
self.current_state = "failed"
elif "password:" in output_lower:
self.current_state = "auth_prompt"
elif "yes/no" in output_lower:
self.current_state = "auth_prompt"
elif "welcome to" in output_lower or "last login" in output_lower:
self.current_state = "connected"
elif output.strip().endswith("$") or output.strip().endswith("#"):
self.current_state = "shell"
elif "connection closed" in output_lower:
self.current_state = "disconnected"
def get_prompt_suggestions(self):
"""Return suggested responses for SSH prompts."""
suggestions = super().get_prompt_suggestions()
if self.current_state == "auth_prompt":
if "password:" in self.multiplexer.get_all_output()["stdout"]:
suggestions.extend(["<password>"])
elif "yes/no" in self.multiplexer.get_all_output()["stdout"]:
suggestions.extend(["yes", "no"])
return suggestions
class GenericProcessHandler(ProcessHandler):
"""Fallback handler for unknown process types."""
def __init__(self, multiplexer):
super().__init__(multiplexer)
self.state_machine = {
"initial": ["running"],
"running": ["waiting_input", "completed"],
"waiting_input": ["running"],
"completed": [],
}
self.prompt_patterns = [
("\\?\\s*$", "waiting_input"),
(">\\s*$", "waiting_input"),
(":\\s*$", "waiting_input"),
("done", "completed"),
("finished", "completed"),
("exit code", "completed"),
]
def get_process_type(self):
return "generic"
def update_state(self, output):
"""Basic state detection for generic processes."""
output_lower = output.lower()
if any((pattern in output_lower for pattern in ["done", "finished", "complete"])):
self.current_state = "completed"
elif any((output.strip().endswith(char) for char in ["?", ">", ":"])):
self.current_state = "waiting_input"
else:
self.current_state = "running"
_handler_classes = {
"apt": AptHandler,
"vim": VimHandler,
"ssh": SSHHandler,
"generic": GenericProcessHandler,
}
def get_handler_for_process(process_type, multiplexer):
"""Get appropriate handler for a process type."""
handler_class = _handler_classes.get(process_type, GenericProcessHandler)
return handler_class(multiplexer)
def detect_process_type(command):
"""Detect process type from command."""
command_str = " ".join(command) if isinstance(command, list) else command
command_lower = command_str.lower()
if "apt" in command_lower or "apt-get" in command_lower:
return "apt"
elif "vim" in command_lower or "vi " in command_lower:
return "vim"
elif "ssh" in command_lower:
return "ssh"
else:
return "generic"
Reference in New Issue View Git Blame Copy Permalink