File Operations

from hopx_ai import Sandbox
import os

API_KEY = os.getenv("HOPX_API_KEY")

with Sandbox.create(template="code-interpreter", api_key=API_KEY) as sandbox:
    # Write text file
    sandbox.files.write("/workspace/hello.txt", "Hello, World!")
    print("File written: /workspace/hello.txt")
    
    # Read file
    content = sandbox.files.read("/workspace/hello.txt")
    print(f"File content: {content}")
    
    # Write binary file
    binary_data = b"\x89PNG\r\n\x1a\n"  # PNG header
    sandbox.files.write_bytes("/workspace/test.png", binary_data)
    print("Binary file written: /workspace/test.png")
    
    # Read binary file
    binary_content = sandbox.files.read_bytes("/workspace/test.png")
    print(f"Binary content length: {len(binary_content)} bytes")

# Create some files
sandbox.files.write("/workspace/file1.txt", "Content 1")
sandbox.files.write("/workspace/file2.txt", "Content 2")
sandbox.files.mkdir("/workspace/subdir")
sandbox.files.write("/workspace/subdir/file3.txt", "Content 3")

# List directory
files = sandbox.files.list("/workspace")
print(f"Files in /workspace: {len(files)}")

for file in files:
    file_type = "📁" if file.is_dir else "📄"
    print(f"  {file_type} {file.name} ({file.size_kb:.2f}KB)")
    print(f"    Path: {file.path}")
    print(f"    Size: {file.size} bytes")
    print(f"    Modified: {file.modified_time}")

import tempfile
import os

# Create a local file to upload
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
    f.write("This is a test file for upload!\nLine 2\nLine 3\n")
    local_file = f.name

# Upload file
sandbox.files.upload(local_file, "/workspace/uploaded.txt")
print(f"File uploaded: {local_file} → /workspace/uploaded.txt")

# Verify upload
content = sandbox.files.read("/workspace/uploaded.txt")
print(f"Uploaded content:\n{content}")

# Download file
download_path = tempfile.mktemp()
sandbox.files.download("/workspace/uploaded.txt", download_path)
print(f"File downloaded: /workspace/uploaded.txt → {download_path}")

# Verify download
with open(download_path, 'r') as f:
    downloaded_content = f.read()
print(f"Downloaded content matches: {content == downloaded_content}")

# Cleanup
os.unlink(local_file)
os.unlink(download_path)

# Check if file exists
exists = sandbox.files.exists("/workspace/hello.txt")
print(f"File exists: {exists}")

# Get file info
if exists:
    files = sandbox.files.list("/workspace")
    file_info = next((f for f in files if f.name == "hello.txt"), None)
    if file_info:
        print(f"Name: {file_info.name}")
        print(f"Size: {file_info.size} bytes")
        print(f"Is directory: {file_info.is_dir}")
        print(f"Modified: {file_info.modified_time}")

# Create directory
sandbox.files.mkdir("/workspace/myproject")
print("Directory created: /workspace/myproject")

# Create nested directories (automatically creates parents)
sandbox.files.mkdir("/workspace/myproject/src")
sandbox.files.mkdir("/workspace/myproject/tests")

# List directory structure
def list_recursive(path, indent=0):
    files = sandbox.files.list(path)
    for file in files:
        print("  " * indent + f"{'📁' if file.is_dir else '📄'} {file.name}")
        if file.is_dir:
            list_recursive(file.path, indent + 1)

list_recursive("/workspace/myproject")

import asyncio

sandbox = Sandbox.create(template="code-interpreter", api_key=API_KEY)

try:
    print("Starting file watcher...")
    
    async def watch_files():
        event_count = 0
        async for event in sandbox.files.watch("/workspace"):
            if event.type == 'change':
                print(f"  📂 {event.event.upper()}: {event.path}")
                event_count += 1
                if event_count >= 5:
                    break
    
    async def make_changes():
        await asyncio.sleep(1)
        await sandbox.files.write("/workspace/test1.txt", "Content 1")
        await asyncio.sleep(0.5)
        await sandbox.files.write("/workspace/test2.txt", "Content 2")
        await asyncio.sleep(0.5)
        await sandbox.files.remove("/workspace/test1.txt")
    
    # Run both concurrently
    await asyncio.gather(watch_files(), make_changes())
    
finally:
    sandbox.kill()

# Remove a file
sandbox.files.remove("/workspace/hello.txt")
print("File removed")

# Remove directory (automatically removes contents)
sandbox.files.remove("/workspace/myproject")
print("Directory removed")

# Verify removal
exists = sandbox.files.exists("/workspace/hello.txt")
print(f"File still exists: {exists}")