mirror of
https://github.com/trycua/computer.git
synced 2026-01-02 03:20:22 -06:00
Merge pull request #439 from trycua/feat/windows-cloud
Support Windows cloud VMs in Computer SDK
This commit is contained in:
ddupont
144
examples/computer_examples_windows.py
Normal file
144
examples/computer_examples_windows.py
Normal file
@@ -0,0 +1,144 @@
|
||||
import os
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import traceback
|
||||
|
||||
# Load environment variables from .env file
|
||||
project_root = Path(__file__).parent.parent
|
||||
env_file = project_root / ".env"
|
||||
print(f"Loading environment from: {env_file}")
|
||||
from computer.helpers import sandboxed
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv(env_file)
|
||||
|
||||
# Add paths to sys.path if needed
|
||||
pythonpath = os.environ.get("PYTHONPATH", "")
|
||||
for path in pythonpath.split(":"):
|
||||
if path and path not in sys.path:
|
||||
sys.path.insert(0, path) # Insert at beginning to prioritize
|
||||
print(f"Added to sys.path: {path}")
|
||||
|
||||
from computer.computer import Computer
|
||||
from computer.providers.base import VMProviderType
|
||||
from computer.logger import LogLevel
|
||||
|
||||
# ANSI color codes
|
||||
RED = '\033[91m'
|
||||
RESET = '\033[0m'
|
||||
|
||||
async def main():
|
||||
try:
|
||||
print("\n=== Using direct initialization ===")
|
||||
|
||||
# Create a remote Windows computer with Cua
|
||||
computer = Computer(
|
||||
os_type="windows",
|
||||
api_key=os.getenv("CUA_API_KEY"),
|
||||
name="s-windows-x4snp46ebf", # os.getenv("CONTAINER_NAME") or "",
|
||||
provider_type=VMProviderType.CLOUD,
|
||||
)
|
||||
|
||||
try:
|
||||
# Run the computer with default parameters
|
||||
await computer.run()
|
||||
|
||||
# Create output directory if it doesn't exist
|
||||
output_dir = Path("./output")
|
||||
output_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Keyboard Actions Examples
|
||||
print("\n=== Keyboard Actions ===")
|
||||
await computer.interface.type_text("Hello, World!")
|
||||
await computer.interface.press_key("enter")
|
||||
|
||||
# Mouse Actions Examples
|
||||
print("\n=== Mouse Actions ===")
|
||||
await computer.interface.move_cursor(100, 100)
|
||||
await computer.interface.left_click()
|
||||
await computer.interface.double_click(400, 400)
|
||||
await computer.interface.right_click(300, 300)
|
||||
|
||||
print("\n=== RPC ===")
|
||||
await computer.venv_install("demo_venv", ["mss"])
|
||||
|
||||
@sandboxed("demo_venv")
|
||||
def greet_and_print(name):
|
||||
from mss import mss
|
||||
import os
|
||||
# get username
|
||||
username = os.getlogin()
|
||||
print(f"Hello from inside the container, {name}!")
|
||||
print("Username:", username)
|
||||
print("Screens:", mss().monitors)
|
||||
|
||||
# take a screenshot
|
||||
with mss() as sct:
|
||||
filename = sct.shot(mon=-1, output='C:/Users/azureuser/Desktop/fullscreen.png')
|
||||
print(filename)
|
||||
|
||||
return {"greeted": name, "username": username}
|
||||
|
||||
# Call with args and kwargs
|
||||
result = await greet_and_print("John Doe")
|
||||
print("Result from sandboxed function:", result)
|
||||
|
||||
# Command Actions Examples
|
||||
print("\n=== Command Actions ===")
|
||||
result = await computer.interface.run_command("notepad")
|
||||
print("Result from command:", result)
|
||||
|
||||
screenshot = await computer.interface.screenshot()
|
||||
screenshot_path = output_dir / "screenshot.png"
|
||||
with open(screenshot_path, "wb") as f:
|
||||
f.write(screenshot)
|
||||
print(f"Screenshot saved to: {screenshot_path.absolute()}")
|
||||
|
||||
# Clipboard Actions Examples
|
||||
print("\n=== Clipboard Actions ===")
|
||||
await computer.interface.set_clipboard("Test clipboard")
|
||||
content = await computer.interface.copy_to_clipboard()
|
||||
print(f"Clipboard content: {content}")
|
||||
|
||||
|
||||
# Simple REPL Loop
|
||||
print("\n=== Command REPL ===")
|
||||
print("Enter commands to run on the remote computer.")
|
||||
print("Type 'exit' or 'quit' to leave the REPL.\n")
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Get command from user
|
||||
command = input("command> ").strip()
|
||||
|
||||
# Check for exit commands
|
||||
if command.lower() in ['exit', 'quit', '']:
|
||||
if command.lower() in ['exit', 'quit']:
|
||||
print("Exiting REPL...")
|
||||
break
|
||||
|
||||
# Run the command
|
||||
result = await computer.interface.run_command(command)
|
||||
|
||||
print(result.stdout)
|
||||
if result.stderr:
|
||||
print(f"{RED}{result.stderr}{RESET}")
|
||||
except KeyboardInterrupt:
|
||||
print("\nExiting REPL...")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"{RED}Error running command: {e}{RESET}")
|
||||
|
||||
|
||||
finally:
|
||||
# Important to clean up resources
|
||||
# await computer.stop()
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"Error in main: {e}")
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -794,19 +794,33 @@ class Computer:
|
||||
Tuple of (stdout, stderr) from the installation command
|
||||
"""
|
||||
requirements = requirements or []
|
||||
|
||||
# Create virtual environment if it doesn't exist
|
||||
venv_path = f"~/.venvs/{venv_name}"
|
||||
create_cmd = f"mkdir -p ~/.venvs && python3 -m venv {venv_path}"
|
||||
|
||||
# Check if venv exists, if not create it
|
||||
check_cmd = f"test -d {venv_path} || ({create_cmd})"
|
||||
_ = await self.interface.run_command(check_cmd)
|
||||
|
||||
# Install packages
|
||||
requirements_str = " ".join(requirements)
|
||||
install_cmd = f". {venv_path}/bin/activate && pip install {requirements_str}"
|
||||
return await self.interface.run_command(install_cmd)
|
||||
# Windows vs POSIX handling
|
||||
if self.os_type == "windows":
|
||||
# Use %USERPROFILE% for home directory and cmd.exe semantics
|
||||
venv_path = f"%USERPROFILE%\\.venvs\\{venv_name}"
|
||||
ensure_dir_cmd = "if not exist \"%USERPROFILE%\\.venvs\" mkdir \"%USERPROFILE%\\.venvs\""
|
||||
create_cmd = f"if not exist \"{venv_path}\" python -m venv \"{venv_path}\""
|
||||
requirements_str = " ".join(requirements)
|
||||
# Activate via activate.bat and install
|
||||
install_cmd = f"call \"{venv_path}\\Scripts\\activate.bat\" && pip install {requirements_str}" if requirements_str else f"echo No requirements to install"
|
||||
await self.interface.run_command(ensure_dir_cmd)
|
||||
await self.interface.run_command(create_cmd)
|
||||
return await self.interface.run_command(install_cmd)
|
||||
else:
|
||||
# POSIX (macOS/Linux)
|
||||
venv_path = f"$HOME/.venvs/{venv_name}"
|
||||
create_cmd = f"mkdir -p \"$HOME/.venvs\" && python3 -m venv \"{venv_path}\""
|
||||
# Check if venv exists, if not create it
|
||||
check_cmd = f"test -d \"{venv_path}\" || ({create_cmd})"
|
||||
_ = await self.interface.run_command(check_cmd)
|
||||
# Install packages
|
||||
requirements_str = " ".join(requirements)
|
||||
install_cmd = (
|
||||
f". \"{venv_path}/bin/activate\" && pip install {requirements_str}"
|
||||
if requirements_str
|
||||
else "echo No requirements to install"
|
||||
)
|
||||
return await self.interface.run_command(install_cmd)
|
||||
|
||||
async def venv_cmd(self, venv_name: str, command: str):
|
||||
"""Execute a shell command in a virtual environment.
|
||||
@@ -818,18 +832,30 @@ class Computer:
|
||||
Returns:
|
||||
Tuple of (stdout, stderr) from the command execution
|
||||
"""
|
||||
venv_path = f"~/.venvs/{venv_name}"
|
||||
|
||||
# Check if virtual environment exists
|
||||
check_cmd = f"test -d {venv_path}"
|
||||
result = await self.interface.run_command(check_cmd)
|
||||
|
||||
if result.stderr or "test:" in result.stdout: # venv doesn't exist
|
||||
return "", f"Virtual environment '{venv_name}' does not exist. Create it first using venv_install."
|
||||
|
||||
# Activate virtual environment and run command
|
||||
full_command = f". {venv_path}/bin/activate && {command}"
|
||||
return await self.interface.run_command(full_command)
|
||||
if self.os_type == "windows":
|
||||
# Windows (cmd.exe)
|
||||
venv_path = f"%USERPROFILE%\\.venvs\\{venv_name}"
|
||||
# Check existence and signal if missing
|
||||
check_cmd = f"if not exist \"{venv_path}\" (echo VENV_NOT_FOUND) else (echo VENV_FOUND)"
|
||||
result = await self.interface.run_command(check_cmd)
|
||||
if "VENV_NOT_FOUND" in getattr(result, "stdout", ""):
|
||||
# Auto-create the venv with no requirements
|
||||
await self.venv_install(venv_name, [])
|
||||
# Activate and run the command
|
||||
full_command = f"call \"{venv_path}\\Scripts\\activate.bat\" && {command}"
|
||||
return await self.interface.run_command(full_command)
|
||||
else:
|
||||
# POSIX (macOS/Linux)
|
||||
venv_path = f"$HOME/.venvs/{venv_name}"
|
||||
# Check if virtual environment exists
|
||||
check_cmd = f"test -d \"{venv_path}\""
|
||||
result = await self.interface.run_command(check_cmd)
|
||||
if result.stderr or "test:" in result.stdout: # venv doesn't exist
|
||||
# Auto-create the venv with no requirements
|
||||
await self.venv_install(venv_name, [])
|
||||
# Activate virtual environment and run command
|
||||
full_command = f". \"{venv_path}/bin/activate\" && {command}"
|
||||
return await self.interface.run_command(full_command)
|
||||
|
||||
async def venv_exec(self, venv_name: str, python_func, *args, **kwargs):
|
||||
"""Execute Python function in a virtual environment using source code extraction.
|
||||
|
||||
Reference in New Issue
Block a user