-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathmorph_provider_example.py
More file actions
199 lines (155 loc) Β· 6.2 KB
/
morph_provider_example.py
File metadata and controls
199 lines (155 loc) Β· 6.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
#!/usr/bin/env python3
"""
Example demonstrating Morph.so provider usage with custom base images and snapshotting.
This example shows:
1. Creating a sandbox with a custom base image
2. Setting up a development environment
3. Creating snapshots for state management
4. Restoring from snapshots
"""
import asyncio
import os
from grainchain.core.interfaces import SandboxConfig
from grainchain.core.sandbox import Sandbox
async def morph_example():
"""Demonstrate Morph provider capabilities."""
# Configuration for Morph provider
# You can customize the base image, resources, etc.
sandbox_config = SandboxConfig(
timeout=300, # 5 minutes
working_directory="~",
provider_config={
"image_id": "morphvm-minimal", # Use minimal base image
"vcpus": 2, # 2 CPU cores
"memory": 2048, # 2GB RAM
"disk_size": 10240, # 10GB disk
},
)
print("π Starting Morph.so sandbox example...")
# Create sandbox using Morph provider
async with Sandbox(provider="morph", config=sandbox_config) as sandbox:
print(f"β
Created Morph sandbox: {sandbox.sandbox_id}")
# 1. Basic setup and testing
print("\nπ¦ Setting up development environment...")
# Install some development tools
result = await sandbox.execute("apt update && apt install -y curl git")
if result.success:
print("β
Installed basic tools")
else:
print(f"β Failed to install tools: {result.stderr}")
return
# Install Python packages
result = await sandbox.execute("pip install requests numpy")
if result.success:
print("β
Installed Python packages")
# 2. Create a simple Python application
print("\nπ Creating a sample application...")
app_code = '''
import requests
import numpy as np
import json
def fetch_data():
"""Fetch some sample data."""
try:
response = requests.get("https://httpbin.org/json")
return response.json()
except Exception as e:
return {"error": str(e)}
def process_data(data):
"""Process data with numpy."""
if "error" in data:
return data
# Create some sample numerical data
arr = np.random.rand(10)
return {
"original": data,
"processed": {
"random_array": arr.tolist(),
"mean": float(np.mean(arr)),
"std": float(np.std(arr))
}
}
if __name__ == "__main__":
print("Fetching and processing data...")
data = fetch_data()
result = process_data(data)
print(json.dumps(result, indent=2))
'''
await sandbox.upload_file("/home/app.py", app_code)
print("β
Created sample application")
# Test the application
result = await sandbox.execute("cd /home && python app.py")
if result.success:
print("β
Application runs successfully")
print(f"π Output preview: {result.stdout[:200]}...")
# 3. Create a snapshot of the configured environment
print("\nπΈ Creating snapshot of configured environment...")
snapshot_id = await sandbox.create_snapshot()
print(f"β
Created snapshot: {snapshot_id}")
# 4. Make some changes to demonstrate snapshot restoration
print("\nπ§ Making changes to the environment...")
# Create some temporary files
await sandbox.execute("echo 'temporary file 1' > /tmp/temp1.txt")
await sandbox.execute("echo 'temporary file 2' > /tmp/temp2.txt")
await sandbox.execute("mkdir -p /tmp/workspace")
# Verify changes exist
result = await sandbox.execute("ls -la /tmp/temp*.txt")
print(f"π Created temporary files: {result.stdout.strip()}")
# 5. Restore from snapshot
print("\nπ Restoring from snapshot...")
try:
await sandbox.restore_snapshot(snapshot_id)
print("β
Restored from snapshot")
# Verify that temporary files are gone but our app remains
result = await sandbox.execute("ls -la /tmp/temp*.txt")
if result.return_code != 0:
print("β
Temporary files were removed (as expected)")
result = await sandbox.execute("ls -la /home/app.py")
if result.success:
print("β
Application file still exists")
except NotImplementedError:
print("βΉοΈ Snapshot restoration not yet implemented for this provider")
# 6. Demonstrate file operations
print("\nπ Testing file operations...")
# List files in home directory
files = await sandbox.list_files("/home")
print(f"π Files in /home: {[f.name for f in files]}")
# Download our application file
content = await sandbox.download_file("/home/app.py")
print(f"π₯ Downloaded app.py ({len(content)} bytes)")
# 7. Performance test
print("\nβ‘ Running performance test...")
perf_script = """
import time
import numpy as np
start = time.time()
# Simulate some computational work
for i in range(100):
arr = np.random.rand(1000)
result = np.fft.fft(arr)
end = time.time()
print(f"Completed 100 FFT operations in {end - start:.2f} seconds")
"""
await sandbox.upload_file("/home/perf_test.py", perf_script)
result = await sandbox.execute("cd /home && python perf_test.py")
if result.success:
print(f"β‘ Performance result: {result.stdout.strip()}")
print("\nπ Morph.so example completed successfully!")
print(f"π Sandbox ID: {sandbox.sandbox_id}")
print(f"π·οΈ Provider: {sandbox.provider_name}")
print(f"πΈ Snapshot ID: {snapshot_id}")
async def main():
"""Main function to run the example."""
try:
await morph_example()
except Exception as e:
print(f"β Example failed: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
# Make sure you have set your Morph API key
if not os.getenv("MORPH_API_KEY"):
print("β οΈ Please set MORPH_API_KEY environment variable")
print(" export MORPH_API_KEY='your-api-key-here'")
exit(1)
asyncio.run(main())