mirror of
https://github.com/trycua/computer.git
synced 2026-05-02 05:01:22 -05:00
Add CommandResult dataclass (#295)
This commit is contained in:
@@ -249,7 +249,8 @@ For complete examples, see [computer_examples.py](./examples/computer_examples.p
|
||||
|
||||
```python
|
||||
# Shell Actions
|
||||
await computer.interface.run_command(cmd) # Run shell command
|
||||
result = await computer.interface.run_command(cmd) # Run shell command
|
||||
# result.stdout, result.stderr, result.returncode
|
||||
|
||||
# Mouse Actions
|
||||
await computer.interface.left_click(x, y) # Left click at coordinates
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional, Dict, Any, Tuple, List
|
||||
from ..logger import Logger, LogLevel
|
||||
from .models import MouseButton
|
||||
from ..models import Computer, CommandResult, MouseButton
|
||||
|
||||
|
||||
class BaseComputerInterface(ABC):
|
||||
@@ -234,8 +234,31 @@ class BaseComputerInterface(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
async def run_command(self, command: str) -> Tuple[str, str]:
|
||||
"""Run shell command."""
|
||||
async def run_command(self, command: str) -> CommandResult:
|
||||
"""Run shell command and return structured result.
|
||||
|
||||
Executes a shell command using subprocess.run with shell=True and check=False.
|
||||
The command is run in the target environment and captures both stdout and stderr.
|
||||
|
||||
Args:
|
||||
command (str): The shell command to execute
|
||||
|
||||
Returns:
|
||||
CommandResult: A structured result containing:
|
||||
- stdout (str): Standard output from the command
|
||||
- stderr (str): Standard error from the command
|
||||
- returncode (int): Exit code from the command (0 indicates success)
|
||||
|
||||
Raises:
|
||||
RuntimeError: If the command execution fails at the system level
|
||||
|
||||
Example:
|
||||
result = await interface.run_command("ls -la")
|
||||
if result.returncode == 0:
|
||||
print(f"Output: {result.stdout}")
|
||||
else:
|
||||
print(f"Error: {result.stderr}, Exit code: {result.returncode}")
|
||||
"""
|
||||
pass
|
||||
|
||||
# Accessibility Actions
|
||||
|
||||
@@ -9,8 +9,7 @@ import websockets
|
||||
from ..logger import Logger, LogLevel
|
||||
from .base import BaseComputerInterface
|
||||
from ..utils import decode_base64_image, encode_base64_image, bytes_to_image, draw_box, resize_image
|
||||
from .models import Key, KeyType, MouseButton
|
||||
|
||||
from ..models import Computer, CommandResult, Key, KeyType, MouseButton
|
||||
|
||||
class LinuxComputerInterface(BaseComputerInterface):
|
||||
"""Interface for Linux."""
|
||||
@@ -616,11 +615,15 @@ class LinuxComputerInterface(BaseComputerInterface):
|
||||
if not result.get("success", False):
|
||||
raise RuntimeError(result.get("error", "Failed to delete directory"))
|
||||
|
||||
async def run_command(self, command: str) -> Tuple[str, str]:
|
||||
async def run_command(self, command: str) -> CommandResult:
|
||||
result = await self._send_command("run_command", {"command": command})
|
||||
if not result.get("success", False):
|
||||
raise RuntimeError(result.get("error", "Failed to run command"))
|
||||
return result.get("stdout", ""), result.get("stderr", "")
|
||||
return CommandResult(
|
||||
stdout=result.get("stdout", ""),
|
||||
stderr=result.get("stderr", ""),
|
||||
returncode=result.get("return_code", 0)
|
||||
)
|
||||
|
||||
# Accessibility Actions
|
||||
async def get_accessibility_tree(self) -> Dict[str, Any]:
|
||||
|
||||
@@ -9,7 +9,7 @@ import websockets
|
||||
from ..logger import Logger, LogLevel
|
||||
from .base import BaseComputerInterface
|
||||
from ..utils import decode_base64_image, encode_base64_image, bytes_to_image, draw_box, resize_image
|
||||
from .models import Key, KeyType, MouseButton
|
||||
from ..models import Computer, CommandResult, Key, KeyType, MouseButton
|
||||
|
||||
|
||||
class MacOSComputerInterface(BaseComputerInterface):
|
||||
@@ -623,11 +623,15 @@ class MacOSComputerInterface(BaseComputerInterface):
|
||||
if not result.get("success", False):
|
||||
raise RuntimeError(result.get("error", "Failed to delete directory"))
|
||||
|
||||
async def run_command(self, command: str) -> Tuple[str, str]:
|
||||
async def run_command(self, command: str) -> CommandResult:
|
||||
result = await self._send_command("run_command", {"command": command})
|
||||
if not result.get("success", False):
|
||||
raise RuntimeError(result.get("error", "Failed to run command"))
|
||||
return result.get("stdout", ""), result.get("stderr", "")
|
||||
return CommandResult(
|
||||
stdout=result.get("stdout", ""),
|
||||
stderr=result.get("stderr", ""),
|
||||
returncode=result.get("return_code", 0)
|
||||
)
|
||||
|
||||
# Accessibility Actions
|
||||
async def get_accessibility_tree(self) -> Dict[str, Any]:
|
||||
|
||||
@@ -9,7 +9,7 @@ import websockets
|
||||
from ..logger import Logger, LogLevel
|
||||
from .base import BaseComputerInterface
|
||||
from ..utils import decode_base64_image, encode_base64_image, bytes_to_image, draw_box, resize_image
|
||||
from .models import Key, KeyType, MouseButton
|
||||
from ..models import Computer, CommandResult, Key, KeyType, MouseButton
|
||||
|
||||
|
||||
class WindowsComputerInterface(BaseComputerInterface):
|
||||
@@ -615,11 +615,15 @@ class WindowsComputerInterface(BaseComputerInterface):
|
||||
if not result.get("success", False):
|
||||
raise RuntimeError(result.get("error", "Failed to delete directory"))
|
||||
|
||||
async def run_command(self, command: str) -> Tuple[str, str]:
|
||||
async def run_command(self, command: str) -> CommandResult:
|
||||
result = await self._send_command("run_command", {"command": command})
|
||||
if not result.get("success", False):
|
||||
raise RuntimeError(result.get("error", "Failed to run command"))
|
||||
return result.get("stdout", ""), result.get("stderr", "")
|
||||
return CommandResult(
|
||||
stdout=result.get("stdout", ""),
|
||||
stderr=result.get("stderr", ""),
|
||||
returncode=result.get("return_code", 0)
|
||||
)
|
||||
|
||||
# Accessibility Actions
|
||||
async def get_accessibility_tree(self) -> Dict[str, Any]:
|
||||
|
||||
@@ -6,6 +6,18 @@ from typing import Optional, Any, Dict
|
||||
# Import base provider interface
|
||||
from .providers.base import BaseVMProvider
|
||||
|
||||
@dataclass
|
||||
class CommandResult:
|
||||
"""Command result."""
|
||||
stdout: str
|
||||
stderr: str
|
||||
returncode: int
|
||||
|
||||
def __init__(self, stdout: str, stderr: str, returncode: int):
|
||||
self.stdout = stdout
|
||||
self.stderr = stderr
|
||||
self.returncode = returncode
|
||||
|
||||
@dataclass
|
||||
class Display:
|
||||
"""Display configuration."""
|
||||
|
||||
Reference in New Issue
Block a user