diff --git a/examples/computer_examples_windows.py b/examples/computer_examples_windows.py new file mode 100644 index 00000000..22c08416 --- /dev/null +++ b/examples/computer_examples_windows.py @@ -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()) \ No newline at end of file diff --git a/libs/python/computer/computer/computer.py b/libs/python/computer/computer/computer.py index f01c5491..6841f2f2 100644 --- a/libs/python/computer/computer/computer.py +++ b/libs/python/computer/computer/computer.py @@ -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.